pub fn add_stream<S>(&self, mut stream: S)
    where
        S: Stream + Unpin + Send + 'static,
        S::Item: 'static + Send,
        A: StreamHandler<S::Item>,
    {
        let mut addr = self.addr.clone();
        task::spawn(async move {
            addr.tx
                .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                    Box::pin(async move {
                        let mut actor = actor.lock().await;
                        StreamHandler::started(&mut *actor, &ctx).await;
                    })
                })))
                .ok();

            while let Some(msg) = stream.next().await {
                if let Err(_) = addr
                    .tx
                    .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                        Box::pin(async move {
                            let mut actor = actor.lock().await;
                            StreamHandler::handle(&mut *actor, &ctx, msg).await;
                        })
                    })))
                {
                    return;
                }
            }

            addr.tx
                .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                    Box::pin(async move {
                        let mut actor = actor.lock().await;
                        StreamHandler::finished(&mut *actor, &ctx).await;
                    })
                })))
                .ok();
        });
    }
/// Create a stream handler for the actor.
///
/// # Examples
/// ```rust
/// use xactor::*;
/// use futures::stream;
/// use async_std::task;
/// use std::time::Duration;
///
/// #[message(result = "i32")]
/// struct GetSum;
///
/// #[derive(Default)]
/// struct MyActor(i32);
///
/// #[async_trait::async_trait]
/// impl StreamHandler<i32> for MyActor {
///     async fn handle(&mut self, _ctx: &Context<Self>, msg: i32) {
///         self.0 += msg;
///     }
///
///     async fn started(&mut self, _ctx: &Context<Self>) {
///         println!("stream started");
///     }
///
///     async fn finished(&mut self, _ctx: &Context<Self>) {
///         println!("stream finished");
///     }
/// }
///
/// #[async_trait::async_trait]
/// impl Handler<GetSum> for MyActor {
///     async fn handle(&mut self, _ctx: &Context<Self>, _msg: GetSum) -> i32 {
///         self.0
///     }
/// }
///
/// #[async_trait::async_trait]
/// impl Actor for MyActor {
///     async fn started(&mut self, ctx: &Context<Self>) {
///         let values = (0..100).collect::<Vec<_>>();
///         ctx.add_stream(stream::iter(values));
///     }
/// }
///
/// #[async_std::main]
/// async fn main() -> Result<()> {
///     let mut addr = MyActor::start_default().await;
///     task::sleep(Duration::from_secs(1)).await; // Wait for the stream to complete
///     let res = addr.call(GetSum).await?;
///     assert_eq!(res, (0..100).sum::<i32>());
///     Ok(())
/// }
/// ```
/// ```
Create a stream handler for the actor.
Create a stream handler for the actor .
pub fn add_stream<S>(&self, mut stream: S)
    where
        S: Stream + Unpin + Send + 'static,
        S::Item: 'static + Send,
        A: StreamHandler<S::Item>,
    {
        let mut addr = self.addr.clone();
        task::spawn(async move {
            addr.tx
                .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                    Box::pin(async move {
                        let mut actor = actor.lock().await;
                        StreamHandler::started(&mut *actor, &ctx).await;
                    })
                })))
                .ok();
            while let Some(msg) = stream.next().await {
                if let Err(_) = addr
                    .tx
                    .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                        Box::pin(async move {
                            let mut actor = actor.lock().await;
                            StreamHandler::handle(&mut *actor, &ctx, msg).await;
                        })
                    })))
                {
                    return;
                }
            }
            addr.tx
                .start_send(ActorEvent::Exec(Box::new(move |actor, ctx| {
                    Box::pin(async move {
                        let mut actor = actor.lock().await;
                        StreamHandler::finished(&mut *actor, &ctx).await;
                    })
                })))
                .ok();
        });
    }
pub fn add_stream < S > ( & self , mut stream : S ) where S : Stream + Unpin + Send + ' static , S :: Item : ' static + Send , A : StreamHandler < S :: Item > , { let mut addr = self . addr . clone ( ) ; task :: spawn ( async move { addr . tx . start_send ( ActorEvent :: Exec ( Box :: new ( move | actor , ctx | { Box :: pin ( async move { let mut actor = actor . lock ( ) . await ; StreamHandler :: started ( & mut * actor , & ctx ) . await ; } ) } ) ) ) . ok ( ) ; while let Some ( msg ) = stream . next ( ) . await { if let Err ( _ ) = addr . tx . start_send ( ActorEvent :: Exec ( Box :: new ( move | actor , ctx | { Box :: pin ( async move { let mut actor = actor . lock ( ) . await ; StreamHandler :: handle ( & mut * actor , & ctx , msg ) . await ; } ) } ) ) ) { return ; } } addr . tx . start_send ( ActorEvent :: Exec ( Box :: new ( move | actor , ctx | { Box :: pin ( async move { let mut actor = actor . lock ( ) . await ; StreamHandler :: finished ( & mut * actor , & ctx ) . await ; } ) } ) ) ) . ok ( ) ; } ) ; }
Create a stream handler for the actor.
Create a stream handler for the actor .
&self
stream
S
&self
stream
S
examples
None
[message(result = "i32")]
None
[derive(default)]
None
[async_trait::async_trait]
impl StreamHandler for MyActor {
async fn handle(&mut self, _ctx: &Context, msg: i32) {
self.0 += msg;
}

async fn started(&mut self, _ctx: &Context) {
println!("stream started");
}

async fn finished(&mut self, _ctx: &Context) {
println!("stream finished");
}
}
impl StreamHandler for MyActor { async fn handle ( &mut self _ctx : &Context msg : i32 ) { self . 0 + = msg ; } async fn started ( &mut self _ctx : &Context ) { println! ( " stream started " ) ; } async fn finished ( &mut self _ctx : &Context ) { println! ( " stream finished " ) ; } }
[async_trait::async_trait]
impl Handler for MyActor {
async fn handle(&mut self, _ctx: &Context, _msg: GetSum) -> i32 {
self.0
}
}
impl Handler for MyActor { async fn handle ( &mut self _ctx : &Context _msg : GetSum ) - > i32 { self . 0 } }
[async_trait::async_trait]
impl Actor for MyActor {
async fn started(&mut self, ctx: &Context) {
let values = (0..100).collect::>();
ctx.add_stream(stream::iter(values));
}
}
impl Actor for MyActor { async fn started ( &mut self ctx : &Context ) { let values = ( 0 .. 100 ) . collect :: > () ; ctx . add_stream ( stream :: iter ( values )) ; } }
[async_std::main]
async fn main() -> Result<()> {
let mut addr = MyActor::start_default().await;
task::sleep(Duration::from_secs(1)).await; // Wait for the stream to complete
let res = addr.call(GetSum).await?;
assert_eq!(res, (0..100).sum::());
Ok(())
}
```
```
async fn main () - > Result< () > { let mut addr = MyActor :: start_default () . await ; task :: sleep ( Duration :: from_secs ( 1 )) . await ; // Wait for the stream to complete let res = addr . call ( GetSum ) . await? ; assert_eq! ( res ( 0 .. 100 ) . sum :: () ) ; Ok (( )) } ` ` ` ` ` `