Mirrored_Repos/AppFlowy
1
0
Fork 0
mirror of https://github.com/AppFlowy-IO/AppFlowy.git synced 2024-08-30 18:12:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

add tracing & test post data to dart

Browse Source
This commit is contained in:
appflowy 2021-07-03 14:14:10 +08:00
parent 1cfcdab124
commit 77aa18d737
34 changed files with 366 additions and 270 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@ -8,6 +8,3 @@ Cargo.lock
# These are backup files generated by rustfmt
**/*.rs.bk
/rust-lib/flowy-derive

10
.idea/libraries/Dart_Packages.xml
View File

@ -524,14 +524,6 @@
</list>
</value>
</entry>
<entry key="protobuf">
<value>
<list>
<option value="$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/protobuf-2.0.0/lib" />
<option value="$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/protobuf-1.1.0/lib" />
</list>
</value>
</entry>
<entry key="provider">
<value>
<list>
@ -841,8 +833,6 @@
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/plugin_platform_interface-2.0.0/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/pool-1.5.0/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/process-4.2.1/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/protobuf-1.1.0/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/protobuf-2.0.0/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/provider-5.0.0/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/pub_semver-2.0.0/lib" />
<root url="file://$PROJECT_DIR$/../../flutter/.pub-cache/hosted/pub.dartlang.org/pubspec_parse-1.0.0/lib" />

1
app_flowy/.vscode/launch.json vendored
View File

@ -9,6 +9,7 @@
"request": "launch",
"program": "${workspaceRoot}/lib/main.dart",
"type": "dart",
// "preLaunchTask": "build rust sdk"
},
{
"name": "app_flowy (profile mode)",

2
app_flowy/lib/startup/startup.dart
View File

@ -25,7 +25,7 @@ class App {
resolveDependencies(env);
// add task
// getIt<AppLauncher>().addTask(RustSDKInitTask());
getIt<AppLauncher>().addTask(RustSDKInitTask());
getIt<AppLauncher>().addTask(AppWidgetTask());
// execute the tasks

21
app_flowy/packages/flowy_sdk/example/pubspec.lock
View File

@ -92,13 +92,6 @@ packages:
url: "https://pub.dartlang.org"
source: hosted
version: "6.1.1"
fixnum:
dependency: transitive
description:
name: fixnum
url: "https://pub.dartlang.org"
source: hosted
version: "0.10.11"
flowy_logger:
dependency: transitive
description:
@ -106,13 +99,6 @@ packages:
relative: true
source: path
version: "0.0.1"
flowy_protobuf:
dependency: transitive
description:
path: "../../flowy_protobuf"
relative: true
source: path
version: "0.0.1"
flowy_sdk:
dependency: "direct main"
description:
@ -215,13 +201,6 @@ packages:
url: "https://pub.dartlang.org"
source: hosted
version: "4.2.1"
protobuf:
dependency: transitive
description:
name: protobuf
url: "https://pub.dartlang.org"
source: hosted
version: "1.1.0"
sky_engine:
dependency: transitive
description: flutter

24
app_flowy/packages/flowy_sdk/lib/ffi/adaptor.dart
View File

@ -1,3 +1,4 @@
import 'dart:convert';
import 'dart:ffi';
// ignore: import_of_legacy_library_into_null_safe
import 'package:isolates/isolates.dart';
@ -5,7 +6,7 @@ import 'package:isolates/isolates.dart';
import 'package:isolates/ports.dart';
import 'package:ffi/ffi.dart';
import 'package:flowy_protobuf/model/grpc.pb.dart';
// ignore: unused_import
import 'package:flutter/services.dart';
import 'dart:async';
import 'dart:typed_data';
@ -22,9 +23,25 @@ class FFIAdaptorException implements Exception {
FFIAdaptorException(this.type);
}
class FFICommand {
final String event;
final Uint8List payload;
FFICommand(this.event, this.payload);
Map<String, dynamic> toJson() => {
'event': event,
'payload': payload,
};
}
class FFIAdaptor {
static Completer<Uint8List> asyncRequest(RequestPacket request) {
Uint8List bytes = request.writeToBuffer();
static Completer<Uint8List> asyncRequest() {
// final command = FFICommand(
// "AuthCheck", Uint8List.fromList(utf8.encode("this is payload")));
final command = FFICommand("AuthCheck", Uint8List(0));
Uint8List bytes = Uint8List.fromList(utf8.encode(jsonEncode(command)));
assert(bytes.isEmpty == false);
if (bytes.isEmpty) {
@ -43,4 +60,3 @@ class FFIAdaptor {
return completer;
}
}

18
app_flowy/packages/flowy_sdk/lib/ffi/ffi.dart
View File

@ -45,25 +45,24 @@ typedef _invoke_async_Dart = void Function(
int len,
);
/// C function `command_sync`.
Pointer<Uint8> sync_command(
Pointer<Uint8> input,
int len,
) {
) {
return _invoke_sync(input, len);
}
final _invoke_sync_Dart _invoke_sync =
_dl.lookupFunction<_invoke_sync_C, _invoke_sync_Dart>('sync_command');
_dl.lookupFunction<_invoke_sync_C, _invoke_sync_Dart>('sync_command');
typedef _invoke_sync_C = Pointer<Uint8> Function(
Pointer<Uint8> input,
Uint64 len,
);
);
typedef _invoke_sync_Dart = Pointer<Uint8> Function(
Pointer<Uint8> input,
int len,
);
);
/// C function `init_sdk`.
int init_sdk(
@ -111,11 +110,12 @@ typedef _store_dart_post_cobject_Dart = void Function(
bool is_tester() {
if (Foundation.kDebugMode) {
// ignore: unnecessary_null_comparison
if (Platform.executable == null) {
// if (Platform.executable.isEmpty) {
// return false;
// } else {
// return Platform.executable.contains("tester");
// }
return false;
} else {
return Platform.executable.contains("tester");
}
} else {
return false;
}

8
app_flowy/packages/flowy_sdk/lib/flowy_sdk.dart
View File

@ -4,6 +4,7 @@ import 'dart:io';
import 'dart:async';
import 'package:flutter/services.dart';
import 'dart:ffi';
import 'ffi/adaptor.dart';
import 'ffi/ffi.dart' as ffi;
import 'package:ffi/ffi.dart';
@ -16,12 +17,13 @@ class FlowySDK {
const FlowySDK();
void dispose() {
}
void dispose() {}
Future<void> init(Directory sdkDir) async {
ffi.store_dart_post_cobject(NativeApi.postCObject);
ffi.init_sdk(sdkDir.path.toNativeUtf8());
final resp = await FFIAdaptor.asyncRequest();
print(resp);
}
}

2
app_flowy/packages/flowy_sdk/macos/Classes/binding.h
View File

@ -8,4 +8,6 @@ int64_t init_sdk(char *path);
void async_command(int64_t port, const uint8_t *input, uintptr_t len);
const uint8_t *sync_command(const uint8_t *input, uintptr_t len);
void link_me_please(void);

14
app_flowy/packages/flowy_sdk/pubspec.lock
View File

@ -204,13 +204,6 @@ packages:
relative: true
source: path
version: "0.0.1"
flowy_protobuf:
dependency: "direct main"
description:
path: "../flowy_protobuf"
relative: true
source: path
version: "0.0.1"
flutter:
dependency: "direct main"
description: flutter
@ -354,13 +347,6 @@ packages:
url: "https://pub.dartlang.org"
source: hosted
version: "1.5.0"
protobuf:
dependency: transitive
description:
name: protobuf
url: "https://pub.dartlang.org"
source: hosted
version: "1.1.0"
pub_semver:
dependency: transitive
description:

2
app_flowy/packages/flowy_sdk/pubspec.yaml
View File

@ -16,8 +16,6 @@ dependencies:
isolates: ^3.0.3+8
flowy_logger:
path: ../flowy_logger
flowy_protobuf:
path: ../flowy_protobuf
infra:
path: ../infra
dartz: '0.10.0-nullsafety.2'

14
app_flowy/pubspec.lock
View File

@ -232,13 +232,6 @@ packages:
relative: true
source: path
version: "0.0.1"
flowy_protobuf:
dependency: "direct main"
description:
path: "packages/flowy_protobuf"
relative: true
source: path
version: "0.0.1"
flowy_sdk:
dependency: "direct main"
description:
@ -501,13 +494,6 @@ packages:
url: "https://pub.dartlang.org"
source: hosted
version: "4.2.1"
protobuf:
dependency: transitive
description:
name: protobuf
url: "https://pub.dartlang.org"
source: hosted
version: "2.0.0"
provider:
dependency: transitive
description:

2
app_flowy/pubspec.yaml
View File

@ -31,8 +31,6 @@ dependencies:
sdk: flutter
flowy_sdk:
path: packages/flowy_sdk
flowy_protobuf:
path: packages/flowy_protobuf
flowy_style:
path: packages/flowy_style

2
rust-lib/dart-ffi/Cargo.toml
View File

@ -20,7 +20,7 @@ byteorder = {version = "1.3.4"}
ffi-support = {version = "0.4.2"}
protobuf = {version = "2.20.0"}
lazy_static = {version = "1.4.0"}
tokio = { version = "1", features = ["sync"] }
tokio = { version = "1", features = ["rt", "rt-multi-thread"] }
log = "0.4.14"
serde = { version = "1.0", features = ["derive"] }
serde_json = {version = "1.0"}

91
rust-lib/dart-ffi/src/lib.rs
View File

@ -3,50 +3,50 @@ mod c;
use crate::c::forget_rust;
use flowy_sdk::*;
use flowy_sys::prelude::*;
use lazy_static::lazy_static;
use std::{cell::RefCell, ffi::CStr, future::Future, os::raw::c_char};
lazy_static! {
pub static ref FFI_RUNTIME: tokio::runtime::Runtime =
tokio::runtime::Builder::new_current_thread()
.thread_name("flowy-dart-ffi")
.build()
.unwrap();
}
#[no_mangle]
pub extern "C" fn init_sdk(path: *mut c_char) -> i64 {
let c_str: &CStr = unsafe { CStr::from_ptr(path) };
let path: &str = c_str.to_str().unwrap();
FlowySDK::init_log();
FlowySDK::init_log(path);
FlowySDK::init(path);
return 1;
}
#[derive(serde::Deserialize)]
pub struct FFICommand {
event: String,
payload: Vec<u8>,
}
impl FFICommand {
pub fn from_bytes(bytes: Vec<u8>) -> Self {
let command: FFICommand = serde_json::from_slice(&bytes).unwrap();
command
}
pub fn from_u8_pointer(pointer: *const u8, len: usize) -> Self {
let bytes = unsafe { std::slice::from_raw_parts(pointer, len) }.to_vec();
FFICommand::from_bytes(bytes)
}
}
#[no_mangle]
pub extern "C" fn async_command(port: i64, input: *const u8, len: usize) {
let FFICommand { event, payload } = FFICommand::from_u8_pointer(input, len);
log::info!("Event: {:?}", event);
let mut request = DispatchRequest::new(port, event).callback(|_, resp| {
log::info!("async resp: {:?}", resp);
});
let mut request = DispatchRequest::new(event);
log::trace!(
"[FFI]: {} Async Event: {:?} with {} port",
&request.id,
&request.event,
port
);
if !payload.is_empty() {
request = request.payload(Payload::Bytes(payload));
}
async_send(request);
spawn_future(async { vec![] }, 123);
request = request.callback(Box::new(move |resp: EventResponse| {
let bytes = match resp.data {
ResponseData::Bytes(bytes) => bytes,
ResponseData::None => vec![],
};
log::trace!("[FFI]: Post data to dart through {} port", port);
Box::pin(spawn_future(async { bytes }, port))
}));
let _ = EventDispatch::async_send(request);
}
#[no_mangle]
@ -56,19 +56,36 @@ pub extern "C" fn sync_command(input: *const u8, len: usize) -> *const u8 { unim
#[no_mangle]
pub extern "C" fn link_me_please() {}
#[derive(serde::Deserialize)]
pub struct FFICommand {
event: String,
payload: Vec<u8>,
}
impl FFICommand {
pub fn from_u8_pointer(pointer: *const u8, len: usize) -> Self {
let bytes = unsafe { std::slice::from_raw_parts(pointer, len) }.to_vec();
let command: FFICommand = serde_json::from_slice(&bytes).unwrap();
command
}
}
#[inline(always)]
fn spawn_future<F>(future: F, port: i64)
async fn spawn_future<F>(future: F, port: i64)
where
F: Future<Output = Vec<u8>> + Send + 'static,
{
let isolate = allo_isolate::Isolate::new(port);
isolate.catch_unwind(future);
// if let Err(e) = isolate.catch_unwind(future) {
// if let Some(msg) = e.downcast_ref::<&str>() {
// log::error!("🔥 {:?}", msg);
// } else {
// log::error!("no info provided for that panic 😡");
// }
// }
match isolate.catch_unwind(future).await {
Ok(success) => {
log::trace!("[FFI]: Post data to dart success");
},
Err(e) => {
if let Some(msg) = e.downcast_ref::<&str>() {
log::error!("[FFI]: ❌ {:?}", msg);
} else {
log::error!("[FFI]: allo_isolate post panic");
}
},
}
}

2
rust-lib/flowy-ast/src/attr.rs
View File

@ -262,7 +262,7 @@ pub struct ASTEnumAttrVariant {
}
impl ASTEnumAttrVariant {
pub fn from_ast(cx: &Ctxt, variant: &syn::Variant) -> Self {
pub fn from_ast(_cx: &Ctxt, variant: &syn::Variant) -> Self {
let name = variant.ident.to_string();
let mut value = String::new();
if variant.discriminant.is_some() {

4
rust-lib/flowy-ast/src/ty_ext.rs
View File

@ -1,5 +1,5 @@
use crate::Ctxt;
use quote::format_ident;
use syn::{self, AngleBracketedGenericArguments, PathSegment};
#[derive(Eq, PartialEq, Debug)]

5
rust-lib/flowy-log/Cargo.toml
View File

@ -6,8 +6,9 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tracing = { version = "0.1" }
tracing-log = { version = "0.1.1", features = ["env_logger"]}
#tracing = { version = "0.1" }
tracing = { version = "0.1", features = ["max_level_debug", "release_max_level_warn"] }
tracing-log = { version = "0.1.1"}
tracing-futures = "0.2.4"
tracing-subscriber = { version = "0.2.12", features = ["registry", "env-filter"] }
tracing-bunyan-formatter = "0.2.2"

79
rust-lib/flowy-log/src/lib.rs
View File

@ -1,37 +1,96 @@
use tracing::subscriber::set_global_default;
use log::LevelFilter;
use std::path::Path;
use tracing::{subscriber::set_global_default, Level};
use tracing_appender::rolling::RollingFileAppender;
use tracing_bunyan_formatter::{BunyanFormattingLayer, JsonStorageLayer};
use tracing_log::LogTracer;
use tracing_subscriber::{layer::SubscriberExt, EnvFilter};
use tracing_subscriber::{
layer::{Layered, SubscriberExt},
EnvFilter,
};
pub fn init_log(name: &str, env_filter: &str) -> std::result::Result<(), String> {
let env_filter =
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(env_filter.to_owned()));
let formatting_layer = BunyanFormattingLayer::new(name.to_owned(), std::io::stdout);
pub struct FlowyLogBuilder {
name: String,
env_filter: String,
directory: String,
file_appender: RollingFileAppender,
}
let subscriber = tracing_subscriber::fmt()
impl FlowyLogBuilder {
pub fn new(name: &str, directory: impl AsRef<Path>) -> Self {
let directory = directory.as_ref().to_str().unwrap().to_owned();
let local_file_name = format!("{}.log", name);
let file_appender = tracing_appender::rolling::hourly(directory.clone(), local_file_name);
FlowyLogBuilder {
name: name.to_owned(),
env_filter: "Info".to_owned(),
directory,
file_appender,
}
}
pub fn env_filter(mut self, env_filter: &str) -> Self {
self.env_filter = env_filter.to_owned();
self
}
pub fn build(self) -> std::result::Result<(), String> {
let env_filter = EnvFilter::new(self.env_filter);
let (non_blocking, _guard) = tracing_appender::non_blocking(self.file_appender);
let formatting_layer = BunyanFormattingLayer::new(self.name, std::io::stdout);
let mut subscriber = tracing_subscriber::fmt()
.with_target(false)
.with_max_level(tracing::Level::TRACE)
.with_writer(std::io::stdout)
.with_thread_ids(false)
.with_target(false)
// .with_writer(non_blocking)
.compact()
.finish()
.with(env_filter)
.with(JsonStorageLayer)
.with(formatting_layer);
let _ = LogTracer::init().map_err(|e| format!("{:?}", e))?;
let _ = LogTracer::builder()
.with_max_level(LevelFilter::Trace)
.init()
.map_err(|e| format!("{:?}", e))
.unwrap();
let _ = set_global_default(subscriber).map_err(|e| format!("{:?}", e))?;
Ok(())
}
}
pub fn init_log(name: &str, directory: &str, env_filter: &str) -> std::result::Result<(), String> {
FlowyLogBuilder::new(name, directory)
.env_filter(env_filter)
.build()
}
#[cfg(test)]
mod tests {
use super::*;
#[derive(Debug)]
struct Position {
x: f32,
y: f32,
}
#[test]
fn test_log() {
init_log("flowy-log", "info").unwrap();
init_log("flowy", ".", "Debug").unwrap();
tracing::info!("😁 Tracing info log");
log::info!("😁 bridge 'log' to 'tracing'");
let pos = Position {
x: 3.234,
y: -1.223,
};
tracing::debug!(?pos.x, ?pos.y);
log::debug!("😁 bridge 'log' to 'tracing'");
}
}

12
rust-lib/flowy-sdk/.gitignore vendored Normal file
View File

@ -0,0 +1,12 @@
# Generated by Cargo
# will have compiled files and executables
/target/
# Remove Cargo.lock from gitignore if creating an executable, leave it for libraries
# More information here https://doc.rust-lang.org/cargo/guide/cargo-toml-vs-cargo-lock.html
Cargo.lock
# These are backup files generated by rustfmt
**/*.rs.bk
**/temp/

4
rust-lib/flowy-sdk/Cargo.toml
View File

@ -6,10 +6,10 @@ edition = "2018"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
flowy-sys = { path = "../flowy-sys" }
flowy-sys = { path = "../flowy-sys", features = ["use_tracing"]}
flowy-log = { path = "../flowy-log" }
flowy-user = { path = "../flowy-user" }
tracing = { version = "0.1" }
log = "0.4.14"
[dev-dependencies]

14
rust-lib/flowy-sdk/src/lib.rs
View File

@ -6,19 +6,17 @@ use module::build_modules;
pub struct FlowySDK {}
impl FlowySDK {
pub fn init_log() { flowy_log::init_log("flowy", "Debug").unwrap(); }
pub fn init_log(directory: &str) { flowy_log::init_log("flowy", directory, "Debug").unwrap(); }
pub fn init(path: &str) {
log::info!("🔥 System start running");
log::debug!("🔥 Root path: {}", path);
log::info!("🔥 Start running");
tracing::info!("🔥 Root path: {}", path);
EventDispatch::construct(|| build_modules());
}
}
pub async fn async_send(data: DispatchRequest<i64>) -> Result<EventResponse, SystemError> {
EventDispatch::async_send(data).await
pub async fn async_send(request: DispatchRequest) -> EventResponse {
EventDispatch::async_send(request).await
}
pub fn sync_send(data: DispatchRequest<i64>) -> Result<EventResponse, SystemError> {
EventDispatch::sync_send(data)
}
pub fn sync_send(request: DispatchRequest) -> EventResponse { EventDispatch::sync_send(request) }

38
rust-lib/flowy-sdk/tests/sdk/helper.rs
View File

@ -3,32 +3,42 @@ pub use flowy_sdk::*;
use flowy_sys::prelude::*;
use std::{
fmt::{Debug, Display},
fs,
hash::Hash,
sync::Once,
};
static INIT: Once = Once::new();
#[allow(dead_code)]
pub fn init_sdk() {
let root_dir = root_dir();
pub fn init_system() {
INIT.call_once(|| {
FlowySDK::init_log();
FlowySDK::init_log(&root_dir);
});
FlowySDK::init("123");
FlowySDK::init(&root_dir);
}
pub struct FlowySDKTester {
request: DispatchRequest<i64>,
fn root_dir() -> String {
let mut path = fs::canonicalize(".").unwrap();
path.push("tests/temp/flowy/");
let path_str = path.to_str().unwrap().to_string();
if !std::path::Path::new(&path).exists() {
std::fs::create_dir_all(path).unwrap();
}
path_str
}
impl FlowySDKTester {
pub struct EventTester {
request: DispatchRequest,
}
impl EventTester {
pub fn new<E>(event: E) -> Self
where
E: Eq + Hash + Debug + Clone + Display,
{
Self {
request: DispatchRequest::new(1, event),
request: DispatchRequest::new(event),
}
}
@ -52,16 +62,18 @@ impl FlowySDKTester {
self
}
#[allow(dead_code)]
pub async fn async_send(self) -> EventResponse {
init_system();
let resp = async_send(self.request).await.unwrap();
init_sdk();
let resp = async_send(self.request).await;
dbg!(&resp);
resp
}
#[allow(dead_code)]
pub fn sync_send(self) -> EventResponse {
init_system();
let resp = sync_send(self.request).unwrap();
init_sdk();
let resp = sync_send(self.request);
dbg!(&resp);
resp
}

10
rust-lib/flowy-sdk/tests/sdk/user_check.rs
View File

@ -4,19 +4,17 @@ use flowy_user::prelude::*;
use tokio::time::{sleep, Duration};
#[test]
#[should_panic]
fn auth_check_no_payload() {
let callback = |_, resp: EventResponse| {
assert_eq!(resp.status, StatusCode::Err);
};
let resp = FlowySDKTester::new(AuthCheck).sync_send();
let resp = EventTester::new(AuthCheck).sync_send();
assert_eq!(resp.status, StatusCode::Ok);
}
#[tokio::test]
async fn auth_check_with_user_name_email_payload() {
let user_data = UserData::new("jack".to_owned(), "helloworld@gmail.com".to_owned());
FlowySDKTester::new(AuthCheck)
EventTester::new(AuthCheck)
.bytes_payload(user_data)
.sync_send();
}

2
rust-lib/flowy-sys/Cargo.toml
View File

@ -28,6 +28,7 @@ serde = { version = "1.0", features = ["derive"] }
#optional crate
bincode = { version = "1.3", optional = true}
protobuf = {version = "2.24.1", optional = true}
tracing = { version = "0.1", optional = true}
[dev-dependencies]
tokio = { version = "1", features = ["full"] }
@ -36,3 +37,4 @@ futures-util = "0.3.15"
[features]
use_serde = ["bincode"]
use_protobuf= ["protobuf"]
use_tracing= ["tracing"]

148
rust-lib/flowy-sys/src/dispatch.rs
View File

@ -8,11 +8,19 @@ use crate::{
};
use derivative::*;
use futures_core::future::BoxFuture;
use futures_util::task::Context;
use lazy_static::lazy_static;
use pin_project::pin_project;
use std::{
fmt::{Debug, Display},
future::Future,
hash::Hash,
sync::RwLock,
thread::JoinHandle,
};
use tokio::{
macros::support::{Pin, Poll},
task::JoinError,
};
lazy_static! {
@ -30,9 +38,9 @@ impl EventDispatch {
F: FnOnce() -> Vec<Module>,
{
let modules = module_factory();
log::debug!("{}", module_info(&modules));
let module_map = as_module_map(modules);
let runtime = tokio_default_runtime().unwrap();
let dispatch = EventDispatch {
module_map,
runtime,
@ -41,115 +49,133 @@ impl EventDispatch {
*(EVENT_DISPATCH.write().unwrap()) = Some(dispatch);
}
pub async fn async_send<T>(request: DispatchRequest<T>) -> Result<EventResponse, SystemError>
where
T: 'static + Debug + Send + Sync,
{
pub fn async_send(request: DispatchRequest) -> DispatchFuture {
match EVENT_DISPATCH.read() {
Ok(dispatch) => {
let dispatch = dispatch.as_ref().unwrap();
let module_map = dispatch.module_map.clone();
let service = Box::new(DispatchService { module_map });
dispatch
.runtime
.spawn(async move { service.call(request).await })
log::trace!("{}: dispatch {:?} to runtime", &request.id, &request.event);
let join_handle = dispatch.runtime.spawn(async move {
service
.call(request)
.await
.unwrap_or_else(|e| {
let msg = format!("{:?}", e);
Ok(InternalError::new(msg).as_response())
.unwrap_or_else(|e| InternalError::new(format!("{:?}", e)).as_response())
});
DispatchFuture {
fut: Box::pin(async move {
join_handle.await.unwrap_or_else(|e| {
InternalError::new(format!("Dispatch join error: {:?}", e))
.as_response()
})
}),
}
},
Err(e) => {
let msg = format!("{:?}", e);
Err(InternalError::new(msg).into())
let msg = format!("Dispatch runtime error: {:?}", e);
log::trace!("{}", msg);
DispatchFuture {
fut: Box::pin(async { InternalError::new(msg).as_response() }),
}
},
}
}
pub fn sync_send<T>(request: DispatchRequest<T>) -> Result<EventResponse, SystemError>
where
T: 'static + Debug + Send + Sync,
{
pub fn sync_send(request: DispatchRequest) -> EventResponse {
futures::executor::block_on(async { EventDispatch::async_send(request).await })
}
}
pub type BoxStreamCallback<T> = Box<dyn FnOnce(T, EventResponse) + 'static + Send + Sync>;
#[pin_project]
pub struct DispatchFuture {
#[pin]
fut: BoxFuture<'static, EventResponse>,
}
impl Future for DispatchFuture {
type Output = EventResponse;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.as_mut().project();
loop {
return Poll::Ready(futures_core::ready!(this.fut.poll(cx)));
}
}
}
pub type BoxFutureCallback =
Box<dyn FnOnce(EventResponse) -> BoxFuture<'static, ()> + 'static + Send + Sync>;
#[derive(Derivative)]
#[derivative(Debug)]
pub struct DispatchRequest<T>
where
T: 'static + Debug,
{
pub config: T,
pub struct DispatchRequest {
pub id: String,
pub event: Event,
pub payload: Option<Payload>,
pub payload: Payload,
#[derivative(Debug = "ignore")]
pub callback: Option<BoxStreamCallback<T>>,
pub callback: Option<BoxFutureCallback>,
}
impl<T> DispatchRequest<T>
where
T: 'static + Debug,
{
pub fn new<E>(config: T, event: E) -> Self
impl DispatchRequest {
pub fn new<E>(event: E) -> Self
where
E: Eq + Hash + Debug + Clone + Display,
{
Self {
config,
payload: None,
payload: Payload::None,
event: event.into(),
id: uuid::Uuid::new_v4().to_string(),
callback: None,
}
}
pub fn payload(mut self, payload: Payload) -> Self {
self.payload = Some(payload);
self.payload = payload;
self
}
pub fn callback<F>(mut self, callback: F) -> Self
where
F: FnOnce(T, EventResponse) + 'static + Send + Sync,
{
self.callback = Some(Box::new(callback));
pub fn callback(mut self, callback: BoxFutureCallback) -> Self {
self.callback = Some(callback);
self
}
pub(crate) fn into_parts(self) -> (ModuleRequest, Option<BoxFutureCallback>) {
let DispatchRequest {
event,
payload,
id,
callback,
} = self;
(ModuleRequest::new(event.clone(), id, payload), callback)
}
}
pub(crate) struct DispatchService {
pub(crate) module_map: ModuleMap,
}
impl<T> Service<DispatchRequest<T>> for DispatchService
where
T: 'static + Debug + Send + Sync,
{
impl Service<DispatchRequest> for DispatchService {
type Response = EventResponse;
type Error = SystemError;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn call(&self, dispatch_request: DispatchRequest<T>) -> Self::Future {
fn call(&self, dispatch_request: DispatchRequest) -> Self::Future {
let module_map = self.module_map.clone();
let DispatchRequest {
config,
event,
payload,
callback,
} = dispatch_request;
let mut request = ModuleRequest::new(event.clone());
if let Some(payload) = payload {
request = request.payload(payload);
};
let (request, callback) = dispatch_request.into_parts();
Box::pin(async move {
let result = {
match module_map.get(&event) {
match module_map.get(&request.event()) {
Some(module) => {
let fut = module.new_service(());
log::trace!(
"{}: handle event: {:?} by {}",
request.id(),
request.event(),
module.name
);
let service_fut = fut.await?.call(request);
service_fut.await
},
@ -158,17 +184,27 @@ where
"Can not find the module to handle the request:{:?}",
request
);
log::trace!("{}", msg);
Err(InternalError::new(msg).into())
},
}
};
let response = result.unwrap_or_else(|e| e.into());
log::trace!("Dispatch result: {:?}", response);
if let Some(callback) = callback {
callback(config, response.clone());
callback(response.clone()).await;
}
Ok(response)
})
}
}
fn module_info(modules: &Vec<Module>) -> String {
let mut info = format!("{} modules loaded\n", modules.len());
for module in modules {
info.push_str(&format!("-> {} loaded \n", module.name));
}
info
}

23
rust-lib/flowy-sys/src/module/module.rs
View File

@ -1,5 +1,6 @@
use std::{
collections::HashMap,
fmt,
fmt::{Debug, Display},
future::Future,
hash::Hash,
@ -53,7 +54,7 @@ impl<T: Display + Eq + Hash + Debug + Clone> std::convert::From<T> for Event {
pub type EventServiceFactory = BoxServiceFactory<(), ServiceRequest, ServiceResponse, SystemError>;
pub struct Module {
name: String,
pub name: String,
data: DataContainer,
service_map: Arc<HashMap<Event, EventServiceFactory>>,
}
@ -112,26 +113,27 @@ pub struct ModuleRequest {
}
impl ModuleRequest {
pub fn new<E>(event: E) -> Self
pub fn new<E>(event: E, id: String, payload: Payload) -> Self
where
E: Into<Event>,
{
Self {
inner: EventRequest::new(event),
payload: Payload::None,
inner: EventRequest::new(event, id),
payload,
}
}
pub fn payload(mut self, payload: Payload) -> Self {
self.payload = payload;
self
}
pub(crate) fn id(&self) -> &str { &self.inner.id }
pub(crate) fn event(&self) -> &Event { &self.inner.event }
}
impl std::fmt::Display for ModuleRequest {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}:{:?}", self.inner.id, self.inner.event)
}
}
impl std::convert::Into<ServiceRequest> for ModuleRequest {
fn into(self) -> ServiceRequest { ServiceRequest::new(self.inner, self.payload) }
}
@ -162,8 +164,9 @@ impl Service<ModuleRequest> for ModuleService {
type Error = SystemError;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
// #[cfg_attr(feature = "use_tracing", xxx)]
#[tracing::instrument(name = "Module Service", level = "debug", skip(self))]
fn call(&self, request: ModuleRequest) -> Self::Future {
log::trace!("Call module service for request {}", &request.id());
match self.service_map.get(&request.event()) {
Some(factory) => {
let service_fut = factory.new_service(());

12
rust-lib/flowy-sys/src/request/request.rs
View File

@ -7,7 +7,6 @@ use crate::{
util::ready::{ready, Ready},
};
use futures_core::ready;
use std::{
fmt::Debug,
@ -23,12 +22,12 @@ pub struct EventRequest {
}
impl EventRequest {
pub fn new<E>(event: E) -> EventRequest
pub fn new<E>(event: E, id: String) -> EventRequest
where
E: Into<Event>,
{
Self {
id: uuid::Uuid::new_v4().to_string(),
id,
event: event.into(),
}
}
@ -63,11 +62,8 @@ impl FromRequest for String {
}
fn unexpected_none_payload(request: &EventRequest) -> SystemError {
log::warn!(
"Event: {:?} expected payload but payload is empty",
&request.event
);
InternalError::new("Expected payload but payload is empty").into()
log::warn!("{:?} expected payload", &request.event);
InternalError::new("Expected payload").into()
}
#[doc(hidden)]

3
rust-lib/flowy-sys/src/response/mod.rs
View File

@ -1,8 +1,9 @@
pub use builder::*;
pub use data::*;
pub use responder::*;
pub use response::*;
mod builder;
pub mod data;
mod data;
mod responder;
mod response;

2
rust-lib/flowy-sys/src/util/mod.rs
View File

@ -6,7 +6,7 @@ pub mod ready;
pub(crate) fn tokio_default_runtime() -> io::Result<tokio::runtime::Runtime> {
runtime::Builder::new_multi_thread()
.thread_name("flowy-sys")
.thread_name("flowy-rt")
.enable_io()
.enable_time()
.on_thread_start(move || {

9
rust-lib/flowy-sys/tests/api/helper.rs
View File

@ -10,14 +10,13 @@ pub fn setup_env() {
});
}
pub async fn async_send(data: DispatchRequest<i64>) -> Result<EventResponse, SystemError> {
EventDispatch::async_send(data).await
pub async fn async_send(request: DispatchRequest) -> EventResponse {
EventDispatch::async_send(request).await
}
pub fn init_system<F>(module_factory: F)
pub fn init_dispatch<F>(module_factory: F)
where
F: FnOnce() -> Vec<Module>,
{
let system = EventDispatch::new(module_factory);
EventDispatch::set_current(system);
EventDispatch::construct(module_factory);
}

8
rust-lib/flowy-sys/tests/api/module.rs
View File

@ -7,9 +7,9 @@ pub async fn hello() -> String { "say hello".to_string() }
async fn test_init() {
setup_env();
let event = "1";
init_system(|| vec![Module::new().event(event, hello)]);
init_dispatch(|| vec![Module::new().event(event, hello)]);
let request = DispatchRequest::new(1, event);
let resp = async_send(request).await.unwrap();
log::info!("sync resp: {:?}", resp);
let request = DispatchRequest::new(event);
let resp = async_send(request).await;
dbg!(&resp);
}

1
rust-lib/flowy-user/src/module.rs
View File

@ -3,6 +3,7 @@ use flowy_sys::prelude::*;
pub fn create() -> Module {
Module::new()
.name("Flowy-User")
.event(AuthCheck, user_check)
.event(SignIn, user_check)
.event(SignUp, user_check)

6
scripts/build_sdk.sh Executable file
View File

@ -0,0 +1,6 @@
#!/bin/sh
#!/usr/bin/env fish
echo 'Start building rust sdk'
rustup show
cargo make desktop