Trait actix::StreamHandler

pub trait StreamHandler<I> where
    Self: Actor, {
    fn handle(&mut self, item: I, ctx: &mut Self::Context);

    fn started(&mut self, ctx: &mut Self::Context) { ... }
    fn finished(&mut self, ctx: &mut Self::Context) { ... }
    fn add_stream<S>(fut: S, ctx: &mut Self::Context) -> SpawnHandle
    where
        Self::Context: AsyncContext<Self>,
        S: Stream<Item = I> + 'static,
        I: 'static,
    { ... }
}

Stream handler

This is helper trait that allows to handle Stream in a similar way as normal actor messages. When stream resolves to a next item, handle() method of this trait get called. If stream produces error, error() method get called. Depends on result of the error() method, actor could continue to process stream items or stop stream processing. When stream completes, finished() method get called. By default finished() method stops actor execution.

Required methods

fn handle(&mut self, item: I, ctx: &mut Self::Context)

Method is called for every message received by this Actor

Loading content...

Provided methods

fn started(&mut self, ctx: &mut Self::Context)

Method is called when stream get polled first time.

fn finished(&mut self, ctx: &mut Self::Context)

Method is called when stream finishes.

By default this method stops actor execution.

fn add_stream<S>(fut: S, ctx: &mut Self::Context) -> SpawnHandle where
    Self::Context: AsyncContext<Self>,
    S: Stream<Item = I> + 'static,
    I: 'static, 

This method register stream to an actor context and allows to handle Stream in similar way as normal actor messages.

use actix::prelude::*;
use futures_util::stream::once;

#[derive(Message)]
#[rtype(result = "()")]
struct Ping;

struct MyActor;

impl StreamHandler<Ping> for MyActor {

    fn handle(&mut self, item: Ping, ctx: &mut Context<MyActor>) {
        println!("PING");
        System::current().stop()
    }

    fn finished(&mut self, ctx: &mut Self::Context) {
        println!("finished");
    }
}

impl Actor for MyActor {
   type Context = Context<Self>;

   fn started(&mut self, ctx: &mut Context<Self>) {
       // add stream
       Self::add_stream(once(async { Ping }), ctx);
   }
}

fn main() {
    let sys = System::new("example");
    let addr = MyActor.start();
    sys.run();
}

Loading content...

Implementors

Loading content...