rename some struct

This commit is contained in:
appflowy 2021-06-29 11:19:12 +08:00
parent 4febb15f08
commit d867093bf5
6 changed files with 130 additions and 130 deletions

View File

@ -18,7 +18,7 @@ pub extern "C" fn async_command(port: i64, input: *const u8, len: usize) {
let bytes = unsafe { std::slice::from_raw_parts(input, len) }.to_vec();
let request = EventRequest::from_data(bytes);
let stream_data = StreamData::new(port, Some(request)).with_callback(Box::new(|_config, response| {
let stream_data = CommandData::new(port, Some(request)).with_callback(Box::new(|_config, response| {
log::info!("async resp: {:?}", response);
}));

View File

@ -19,14 +19,14 @@ pub fn init_system<F>(modules: Vec<Module>) {
FlowySystem::construct(
|| modules,
|module_map| {
let mut stream = CommandStream::<i64>::new(module_map.clone());
let stream_fut = CommandStreamFuture::new(module_map, stream.take_data_rx());
let mut stream = CommandSender::<i64>::new(module_map.clone());
let runner = CommandSenderRunner::new(module_map, stream.take_data_rx());
STREAM_SENDER.with(|cell| {
CMD_SENDER.with(|cell| {
*cell.borrow_mut() = Some(stream);
});
stream_fut
runner
},
)
.run()
@ -34,18 +34,18 @@ pub fn init_system<F>(modules: Vec<Module>) {
}
thread_local!(
static STREAM_SENDER: RefCell<Option<CommandStream<i64>>> = RefCell::new(None);
static CMD_SENDER: RefCell<Option<CommandSender<i64>>> = RefCell::new(None);
);
pub fn sync_send(data: StreamData<i64>) -> EventResponse {
STREAM_SENDER.with(|cell| match &*cell.borrow() {
pub fn sync_send(data: CommandData<i64>) -> EventResponse {
CMD_SENDER.with(|cell| match &*cell.borrow() {
Some(stream) => stream.sync_send(data),
None => panic!(""),
})
}
pub fn async_send(data: StreamData<i64>) {
STREAM_SENDER.with(|cell| match &*cell.borrow() {
pub fn async_send(data: CommandData<i64>) {
CMD_SENDER.with(|cell| match &*cell.borrow() {
Some(stream) => {
stream.async_send(data);
},

View File

@ -15,131 +15,35 @@ use tokio::{
macro_rules! service_factor_impl {
($name:ident) => {
#[allow(non_snake_case, missing_docs)]
impl<T> ServiceFactory<StreamData<T>> for $name<T>
impl<T> ServiceFactory<CommandData<T>> for $name<T>
where
T: 'static,
{
type Response = EventResponse;
type Error = SystemError;
type Service = BoxService<StreamData<T>, Self::Response, Self::Error>;
type Service = BoxService<CommandData<T>, Self::Response, Self::Error>;
type Config = ();
type Future = LocalBoxFuture<'static, Result<Self::Service, Self::Error>>;
fn new_service(&self, _cfg: Self::Config) -> Self::Future {
let module_map = self.module_map.clone();
let service = Box::new(CommandStreamService { module_map });
let service = Box::new(CommandSenderService { module_map });
Box::pin(async move { Ok(service as Self::Service) })
}
}
};
}
pub type BoxStreamCallback<T> = Box<dyn FnOnce(T, EventResponse) + 'static + Send + Sync>;
pub struct StreamData<T>
where
T: 'static,
{
config: T,
request: Option<EventRequest>,
callback: Option<BoxStreamCallback<T>>,
}
impl<T> StreamData<T> {
pub fn new(config: T, request: Option<EventRequest>) -> Self {
Self {
config,
request,
callback: None,
}
}
pub fn with_callback(mut self, callback: BoxStreamCallback<T>) -> Self {
self.callback = Some(callback);
self
}
}
pub struct CommandStream<T>
where
T: 'static,
{
module_map: ModuleMap,
data_tx: UnboundedSender<StreamData<T>>,
data_rx: Option<UnboundedReceiver<StreamData<T>>>,
}
service_factor_impl!(CommandStream);
impl<T> CommandStream<T> {
pub fn new(module_map: ModuleMap) -> Self {
let (data_tx, data_rx) = unbounded_channel::<StreamData<T>>();
Self {
module_map,
data_tx,
data_rx: Some(data_rx),
}
}
pub fn async_send(&self, data: StreamData<T>) { let _ = self.data_tx.send(data); }
pub fn sync_send(&self, data: StreamData<T>) -> EventResponse {
let factory = self.new_service(());
futures::executor::block_on(async {
let service = factory.await.unwrap();
service.call(data).await.unwrap()
})
}
pub fn tx(&self) -> UnboundedSender<StreamData<T>> { self.data_tx.clone() }
pub fn take_data_rx(&mut self) -> UnboundedReceiver<StreamData<T>> { self.data_rx.take().unwrap() }
}
pub struct CommandStreamFuture<T: 'static> {
module_map: ModuleMap,
data_rx: UnboundedReceiver<StreamData<T>>,
}
service_factor_impl!(CommandStreamFuture);
impl<T: 'static> CommandStreamFuture<T> {
pub fn new(module_map: ModuleMap, data_rx: UnboundedReceiver<StreamData<T>>) -> Self {
Self { module_map, data_rx }
}
}
impl<T> Future for CommandStreamFuture<T>
where
T: 'static,
{
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
match ready!(Pin::new(&mut self.data_rx).poll_recv(cx)) {
None => return Poll::Ready(()),
Some(ctx) => {
let factory = self.new_service(());
tokio::task::spawn_local(async move {
let service = factory.await.unwrap();
let _ = service.call(ctx).await;
});
},
}
}
}
}
pub struct CommandStreamService {
struct CommandSenderService {
module_map: ModuleMap,
}
impl<T: 'static> Service<StreamData<T>> for CommandStreamService {
impl<T: 'static> Service<CommandData<T>> for CommandSenderService {
type Response = EventResponse;
type Error = SystemError;
type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;
fn call(&self, mut data: StreamData<T>) -> Self::Future {
fn call(&self, mut data: CommandData<T>) -> Self::Future {
let module_map = self.module_map.clone();
let request = data.request.take().unwrap();
let fut = async move {
@ -168,3 +72,99 @@ impl<T: 'static> Service<StreamData<T>> for CommandStreamService {
Box::pin(fut)
}
}
pub type BoxStreamCallback<T> = Box<dyn FnOnce(T, EventResponse) + 'static + Send + Sync>;
pub struct CommandData<T>
where
T: 'static,
{
config: T,
request: Option<EventRequest>,
callback: Option<BoxStreamCallback<T>>,
}
impl<T> CommandData<T> {
pub fn new(config: T, request: Option<EventRequest>) -> Self {
Self {
config,
request,
callback: None,
}
}
pub fn with_callback(mut self, callback: BoxStreamCallback<T>) -> Self {
self.callback = Some(callback);
self
}
}
pub struct CommandSender<T>
where
T: 'static,
{
module_map: ModuleMap,
data_tx: UnboundedSender<CommandData<T>>,
data_rx: Option<UnboundedReceiver<CommandData<T>>>,
}
service_factor_impl!(CommandSender);
impl<T> CommandSender<T> {
pub fn new(module_map: ModuleMap) -> Self {
let (data_tx, data_rx) = unbounded_channel::<CommandData<T>>();
Self {
module_map,
data_tx,
data_rx: Some(data_rx),
}
}
pub fn async_send(&self, data: CommandData<T>) { let _ = self.data_tx.send(data); }
pub fn sync_send(&self, data: CommandData<T>) -> EventResponse {
let factory = self.new_service(());
futures::executor::block_on(async {
let service = factory.await.unwrap();
service.call(data).await.unwrap()
})
}
pub fn tx(&self) -> UnboundedSender<CommandData<T>> { self.data_tx.clone() }
pub fn take_data_rx(&mut self) -> UnboundedReceiver<CommandData<T>> { self.data_rx.take().unwrap() }
}
pub struct CommandSenderRunner<T: 'static> {
module_map: ModuleMap,
data_rx: UnboundedReceiver<CommandData<T>>,
}
service_factor_impl!(CommandSenderRunner);
impl<T: 'static> CommandSenderRunner<T> {
pub fn new(module_map: ModuleMap, data_rx: UnboundedReceiver<CommandData<T>>) -> Self {
Self { module_map, data_rx }
}
}
impl<T> Future for CommandSenderRunner<T>
where
T: 'static,
{
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
loop {
match ready!(Pin::new(&mut self.data_rx).poll_recv(cx)) {
None => return Poll::Ready(()),
Some(ctx) => {
let factory = self.new_service(());
tokio::task::spawn_local(async move {
let service = factory.await.unwrap();
let _ = service.call(ctx).await;
});
},
}
}
}
}

View File

@ -1,7 +1,7 @@
use crate::{
module::{Event, Module},
rt::Runtime,
stream::CommandStreamFuture,
stream::CommandSenderRunner,
};
use futures_core::{ready, task::Context};
use std::{cell::RefCell, collections::HashMap, future::Future, io, rc::Rc, sync::Arc};
@ -28,10 +28,10 @@ pub struct FlowySystem {
}
impl FlowySystem {
pub fn construct<F, S, T>(module_factory: F, stream_factory: S) -> SystemRunner
pub fn construct<F, S, T>(module_factory: F, sender_factory: S) -> SystemRunner
where
F: FnOnce() -> Vec<Module>,
S: FnOnce(ModuleMap) -> CommandStreamFuture<T>,
S: FnOnce(ModuleMap) -> CommandSenderRunner<T>,
T: 'static,
{
let runtime = Runtime::new().unwrap();
@ -54,8 +54,8 @@ impl FlowySystem {
});
let system = Self { sys_cmd_tx };
let stream_fut = stream_factory(Rc::new(module_map));
runtime.spawn(stream_fut);
let sender_runner = sender_factory(Rc::new(module_map));
runtime.spawn(sender_runner);
FlowySystem::set_current(system);
let runner = SystemRunner { rt: runtime, stop_rx };

View File

@ -1,4 +1,4 @@
use flowy_sys::prelude::{CommandStream, CommandStreamFuture, EventResponse, FlowySystem, Module, StreamData};
use flowy_sys::prelude::{CommandData, CommandSender, CommandSenderRunner, EventResponse, FlowySystem, Module};
use std::{
cell::RefCell,
sync::{Once, RwLock},
@ -21,18 +21,18 @@ pub struct ExecutorAction {
pub struct FlowySystemExecutor {}
thread_local!(
static STREAM_SENDER: RefCell<Option<CommandStream<i64>>> = RefCell::new(None);
static CMD_SENDER: RefCell<Option<CommandSender<i64>>> = RefCell::new(None);
);
pub fn sync_send(data: StreamData<i64>) -> EventResponse {
STREAM_SENDER.with(|cell| match &*cell.borrow() {
pub fn sync_send(data: CommandData<i64>) -> EventResponse {
CMD_SENDER.with(|cell| match &*cell.borrow() {
Some(stream) => stream.sync_send(data),
None => panic!(""),
})
}
pub fn async_send(data: StreamData<i64>) {
STREAM_SENDER.with(|cell| match &*cell.borrow() {
pub fn async_send(data: CommandData<i64>) {
CMD_SENDER.with(|cell| match &*cell.borrow() {
Some(stream) => {
stream.async_send(data);
},
@ -40,8 +40,6 @@ pub fn async_send(data: StreamData<i64>) {
});
}
pub fn stop_system() { FlowySystem::current().stop(); }
pub fn init_system<F>(modules: Vec<Module>, f: F)
where
F: FnOnce() + 'static,
@ -49,17 +47,19 @@ where
FlowySystem::construct(
|| modules,
|module_map| {
let mut stream = CommandStream::<i64>::new(module_map.clone());
let stream_fut = CommandStreamFuture::new(module_map, stream.take_data_rx());
let mut stream = CommandSender::<i64>::new(module_map.clone());
let runner = CommandSenderRunner::new(module_map, stream.take_data_rx());
STREAM_SENDER.with(|cell| {
CMD_SENDER.with(|cell| {
*cell.borrow_mut() = Some(stream);
});
stream_fut
runner
},
)
.spawn(async { f() })
.run()
.unwrap();
}
pub fn stop_system() { FlowySystem::current().stop(); }

View File

@ -20,7 +20,7 @@ fn test_init() {
init_system(modules, || {
let request = EventRequest::new(no_params_command);
let stream_data = StreamData::new(1, Some(request)).with_callback(Box::new(|_config, response| {
let stream_data = CommandData::new(1, Some(request)).with_callback(Box::new(|_config, response| {
log::info!("async resp: {:?}", response);
}));