Implement ScopeExt for Reactor.

packages/yew-agent/src/lib.rs

@@ -96,13 +96,13 @@ pub mod prelude {
     //! Prelude module to be imported when working with `yew-agent`.
     //!
     //! This module re-exports the frequently used types from the crate.
-    pub use crate::oneshot::{oneshot, use_bridge_oneshot, UseBridgeOneshotHandle};
+    pub use crate::oneshot::{oneshot, use_oneshot_bridge, UseOneshotBridgeHandle};
     pub use crate::reach::Reach;
     pub use crate::reactor::{
-        use_reactor_bridge, use_reactor_subscription, ReactorEvent, UseReactorBridgeHandle,
-        UseReactorSubscriptionHandle,
+        reactor, use_reactor_bridge, use_reactor_subscription, ReactorEvent,
+        UseReactorBridgeHandle, UseReactorSubscriptionHandle,
     };
-    pub use crate::scope_ext::{AgentScopeExt, /* ReactorBridgeHandle, */ WorkerBridgeHandle};
+    pub use crate::scope_ext::{AgentScopeExt, ReactorBridgeHandle, WorkerBridgeHandle};
     pub use crate::worker::{
         use_worker_bridge, use_worker_subscription, UseWorkerBridgeHandle,
         UseWorkerSubscriptionHandle,

packages/yew-agent/src/oneshot/hooks.rs

@@ -3,16 +3,16 @@ use yew::prelude::*;
 use super::provider::OneshotProviderState;
 use super::Oneshot;
 
-/// Handle for [use_bridge_oneshot]
+/// Handle for [use_oneshot_bridge]
 #[derive(Debug)]
-pub struct UseBridgeOneshotHandle<T>
+pub struct UseOneshotBridgeHandle<T>
 where
     T: Oneshot + 'static,
 {
     state: OneshotProviderState<T>,
 }
 
-impl<T> UseBridgeOneshotHandle<T>
+impl<T> UseOneshotBridgeHandle<T>
 where
     T: Oneshot + 'static,
 {
@@ -22,7 +22,7 @@ where
     }
 }
 
-impl<T> Clone for UseBridgeOneshotHandle<T>
+impl<T> Clone for UseOneshotBridgeHandle<T>
 where
     T: Oneshot + 'static,
 {
@@ -33,7 +33,7 @@ where
     }
 }
 
-impl<T> PartialEq for UseBridgeOneshotHandle<T>
+impl<T> PartialEq for UseOneshotBridgeHandle<T>
 where
     T: Oneshot,
 {
@@ -44,11 +44,11 @@ where
 
 /// A hook to bridge to an oneshot agent.
 #[hook]
-pub fn use_bridge_oneshot<T>() -> UseBridgeOneshotHandle<T>
+pub fn use_oneshot_bridge<T>() -> UseOneshotBridgeHandle<T>
 where
     T: Oneshot + 'static,
 {
     let state = use_context::<OneshotProviderState<T>>().expect("failed to find worker context");
 
-    UseBridgeOneshotHandle { state }
+    UseOneshotBridgeHandle { state }
 }

packages/yew-agent/src/oneshot/mod.rs

@@ -5,7 +5,7 @@ mod provider;
 
 #[doc(inline)]
 pub use gloo_worker::oneshot::{Oneshot, OneshotBridge, OneshotRegistrar, OneshotSpawner};
-pub use hooks::{use_bridge_oneshot, UseBridgeOneshotHandle};
+pub use hooks::{use_oneshot_bridge, UseOneshotBridgeHandle};
 pub use provider::OneshotProvider;
 pub(crate) use provider::OneshotProviderState;
 /// A procedural macro to create oneshot agents.

packages/yew-agent/src/reactor/mod.rs

@@ -17,8 +17,7 @@
 //! # #[derive(Serialize, Deserialize)]
 //! # pub struct ReactorOutput {}
 //! #
-//! use yew_agent::reactor;
-//! use yew_agent::reactor::{ReactorReceiver, ReactorSender};
+//! use yew_agent::reactor::{reactor, ReactorReceiver, ReactorSender};
 //! #[reactor(MyReactor)]
 //! pub async fn my_reactor(rx: ReactorReceiver<ReactorInput>, tx: ReactorSender<ReactorOutput>) {
 //!     while let Some(input) = rx.next().await {
@@ -47,5 +46,6 @@ pub use hooks::{
     UseReactorSubscriptionHandle,
 };
 pub use provider::ReactorProvider;
+pub(crate) use provider::ReactorProviderState;
 /// A procedural macro to create reactor agents.
 pub use yew_agent_macro::reactor;

packages/yew-agent/src/scope_ext.rs

@@ -1,14 +1,18 @@
 //! This module contains extensions to the component scope for agent access.
 
+use std::fmt;
+use std::rc::Rc;
+
+use futures::stream::SplitSink;
+use futures::{SinkExt, StreamExt};
 use wasm_bindgen::UnwrapThrowExt;
 use yew::html::Scope;
+use yew::platform::pinned::RwLock;
 use yew::platform::spawn_local;
 use yew::prelude::*;
 
-// use crate::reactor::{
-//     Reactor, ReactorInput, ReactorOutput, ReactorReceivable, ReactorSendable, ReactorWorker,
-// };
 use crate::oneshot::{Oneshot, OneshotProviderState};
+use crate::reactor::{Reactor, ReactorBridge, ReactorEvent, ReactorProviderState, ReactorScoped};
 use crate::worker::{Worker, WorkerBridge, WorkerProviderState};
 
 /// A Worker Bridge Handle.
@@ -30,24 +34,40 @@ where
     }
 }
 
-// /// A Reactor Bridge Handle.
-// #[derive(Debug)]
-// pub struct ReactorBridgeHandle<R>
-// where
-//     R: Reactor + 'static,
-// {
-//     inner: WorkerBridge<ReactorWorker<R>>,
-// }
-
-// impl<R> ReactorBridgeHandle<R>
-// where
-//     R: Reactor + 'static,
-// {
-//     /// Sends a message to the reactor agent.
-//     pub fn send(&self, input: <R::Receiver as ReactorReceivable>::Input) {
-//         self.inner.send(ReactorInput::Input(input))
-//     }
-// }
+type ReactorTx<R> =
+    Rc<RwLock<SplitSink<ReactorBridge<R>, <<R as Reactor>::Scope as ReactorScoped>::Input>>>;
+
+/// A Reactor Bridge Handle.
+pub struct ReactorBridgeHandle<R>
+where
+    R: Reactor + 'static,
+{
+    tx: ReactorTx<R>,
+}
+
+impl<R> fmt::Debug for ReactorBridgeHandle<R>
+where
+    R: Reactor + 'static,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ReactorBridgeHandle<_>")
+            .finish_non_exhaustive()
+    }
+}
+
+impl<R> ReactorBridgeHandle<R>
+where
+    R: Reactor + 'static,
+{
+    /// Sends a message to the reactor agent.
+    pub fn send(&self, input: <R::Scope as ReactorScoped>::Input) {
+        let tx = self.tx.clone();
+        spawn_local(async move {
+            let mut tx = tx.write().await;
+            let _ = tx.send(input).await;
+        });
+    }
+}
 
 /// An extension to [`Scope`](yew::html::Scope) that provides communication mechanism to agents.
 ///
@@ -58,14 +78,11 @@ pub trait AgentScopeExt {
     where
         W: Worker + 'static;
 
-    // /// Bridges to a Reactor Agent.
-    // fn bridge_reactor<R>(
-    //     &self,
-    //     callback: Callback<ReactorOutput<<R::Sender as ReactorSendable>::Output>>,
-    // ) -> ReactorBridgeHandle<R>
-    // where
-    //     R: Reactor + 'static,
-    //     <R::Sender as ReactorSendable>::Output: 'static;
+    /// Bridges to a Reactor Agent.
+    fn bridge_reactor<R>(&self, callback: Callback<ReactorEvent<R>>) -> ReactorBridgeHandle<R>
+    where
+        R: Reactor + 'static,
+        <R::Scope as ReactorScoped>::Output: 'static;
 
     /// Runs a task in a Task Agent.
     fn run_task<T>(&self, input: T::Input, callback: Callback<T::Output>)
@@ -90,18 +107,30 @@ where
         WorkerBridgeHandle { inner }
     }
 
-    // fn bridge_reactor<R>(
-    //     &self,
-    //     callback: Callback<ReactorOutput<<R::Sender as ReactorSendable>::Output>>,
-    // ) -> ReactorBridgeHandle<R>
-    // where
-    //     R: Reactor,
-    //     <R::Sender as ReactorSendable>::Output: 'static,
-    // {
-    //     let inner = self.bridge_worker::<ReactorWorker<R>>(callback).inner;
-
-    //     ReactorBridgeHandle { inner }
-    // }
+    fn bridge_reactor<R>(&self, callback: Callback<ReactorEvent<R>>) -> ReactorBridgeHandle<R>
+    where
+        R: Reactor + 'static,
+        <R::Scope as ReactorScoped>::Output: 'static,
+    {
+        let (tx, mut rx) = self
+            .context::<ReactorProviderState<R>>((|_| {}).into())
+            .expect_throw("failed to bridge to agent.")
+            .0
+            .create_bridge()
+            .split();
+
+        spawn_local(async move {
+            while let Some(m) = rx.next().await {
+                callback.emit(ReactorEvent::<R>::Output(m));
+            }
+
+            callback.emit(ReactorEvent::<R>::Finished);
+        });
+
+        let tx = Rc::new(RwLock::new(tx));
+
+        ReactorBridgeHandle { tx }
+    }
 
     fn run_task<T>(&self, input: T::Input, callback: Callback<T::Output>)
     where