Trait actix::prelude::StreamHandler

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

    fn started(&mut self, ctx: &mut Self::Context) { ... }
    fn error(&mut self, err: E, ctx: &mut Self::Context) -> Running { ... }
    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, Error = E> + 'static,
        I: 'static,
        E: '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

Provided Methods

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

Method is called when stream get polled first time.

fn error(&mut self, err: E, ctx: &mut Self::Context) -> Running

Method is called when stream emits error.

If this method returns ErrorAction::Continue stream processing continues otherwise stream processing stops. Default method implementation returns ErrorAction::Stop

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, Error = E> + 'static,
    I: 'static,
    E: '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::stream::once;

#[derive(Message)]
struct Ping;

struct MyActor;

impl StreamHandler<Ping, io::Error> for MyActor {

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

    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::<Ping, io::Error>(Ok(Ping)), ctx);
   }
}

Implementors