Actix actor의 AsyncContext 활용하기 (2)

지난 글에 이어 Actix actor의 AsyncContext 함수를 사용하는 방법을 알아봅니다.

add_stream, add_message_stream

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


fn add_stream<S>(&mut self, fut: S) -> SpawnHandle
where
    S: Stream + 'static,
    A: StreamHandler<S::Item>;

fn add_message_stream<S>(&mut self, fut: S)
    where
        S: Stream + 'static,
        S::Item: Message,
        A: Handler<S::Item>;

Context에 stream을 등록하는 함수입니다. 두 함수는 거의 비슷하나 차이점은 add_message_stream은 stream error를 무시한다는 점입니다.

다음은 API Doc의 예제인데요. Stream을 만들고 message를 하나 보냅니다. Stream의 사용법과 예제는 다음 글에서 따로 정리해 보겠습니다 :)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35


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
        ctx.add_stream(once(async { Ping }));
    }
}

fn main() {
    let mut sys = System::new();
    let addr = sys.block_on(async { MyActor.start() });
    sys.run();
}

실행 결과는 다음과 같습니다.

1
2


PING
finished

notify, notify_later

1
2
3
4
5
6
7
8
9


fn notify<M>(&mut self, msg: M)
where
    A: Handler<M>,
    M: Message + 'static;

fn notify_later<M>(&mut self, msg: M, after: Duration) -> SpawnHandle
    where
        A: Handler<M>,
        M: Message + 'static;

notify는 자기 자신에게 message를 보내는 함수입니다. 이전 글에서 address를 통해 자기 자신의 주소를 찾고, 여기에 message를 보내는 예제를 작성하였는데요. notify 함수로 바로 message handler를 부를 수도 있습니다. 이 함수는 actor의 mailbox를 bypass 하고 항상 message를 전달한다고 합니다. 하지만 actor가 Stopped 상태라면 error 발생합니다.

이전에 자신의 Addr를 찾아 메시지를 보내는 예제를 notify 함수를 사용하도록 수정하면 다음과 같습니다.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


fn send_message_to_myself_using_address(&mut self, ctx: &mut Context<Self>) {
    let my_address = ctx.address();
    actix::spawn(async move {
        my_address.send(Message).await;
    });
}

fn send_message_to_myself_using_notify(&mut self, ctx: &mut Context<Self>) {
    ctx.notify(Message);
}

notify_later은 notify 함수와 유사하지만, 지정된 시간 이후에 message를 보냅니다. 또 하나의 차이점은 SpawnHandle을 리턴한다는 것 인데요. 실행되기 전에는 cancel_future 통해 동작을 취소할 수 있습니다. 이전 글에서 언급한 바와 같이, cancel_future 함수의 리턴 값은 항상 true 입니다. 취소 성공 여부를 리턴하지 않음을 참고하세요.

1
2
3
4


fn send_message_to_myself_and_cancel(&mut self, ctx: &mut Context<Self>) {
    let handle = ctx.notify(Message, Duration::from_secs(3));
    let _success = ctx.cancel_future(handle);
}

run_later, run_interval

1
2
3
4
5
6
7


fn run_later<F>(&mut self, dur: Duration, f: F) -> SpawnHandle
where
    F: FnOnce(&mut A, &mut A::Context) + 'static;

fn run_interval<F>(&mut self, dur: Duration, f: F) -> SpawnHandle
    where
        F: FnMut(&mut A, &mut A::Context) + 'static;

run_later은 주어진 시간 이후에 주어진 future를 실행시키고, run_interval은 주어진 시간 주기로 future를 실행시킵니다. 사용법은 거의 동일하고, run_interval의 사용 예제는 이 글을 참고하시면 되겠습니다.

References

https://docs.rs/actix/0.13.0/actix/trait.AsyncContext.html

add_stream, add_message_stream#

notify, notify_later#

run_later, run_interval#

References#

add_stream, add_message_stream

notify, notify_later

run_later, run_interval

References