meio 0.95.2

A compact actors framework for the async epoch in Rust & Tokio
Documentation 
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
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

use crate::actor_runtime::Context;
use crate::handlers::{Consumer, InstantAction, InstantActionHandler, StreamAcceptor, StreamItem};
use crate::linkage::ActionRecipient;
use crate::lite_runtime::{LiteTask, Tag};
use anyhow::Error;
use async_trait::async_trait;
use futures::{Stream, StreamExt};
use uuid::Uuid;

/// This worker receives items from a stream and send them as actions
/// into an `Actor`.
pub(crate) struct StreamForwarder<S: Stream> {
    stream: S,
    recipient: Box<dyn ActionRecipient<StreamItem<S::Item>>>,
}

/// If you need chunks use `ready_chunk` on the stream before.
impl<S> StreamForwarder<S>
where
    S: Stream,
    S::Item: Send + 'static,
{
    pub fn new(stream: S, recipient: impl ActionRecipient<StreamItem<S::Item>>) -> Self {
        Self {
            stream,
            recipient: Box::new(recipient),
        }
    }
}

#[async_trait]
impl<S> LiteTask for StreamForwarder<S>
where
    S: Stream + Unpin + Send + 'static,
    S::Item: Send + 'static,
{
    type Output = ();

    fn name(&self) -> String {
        let uuid = Uuid::new_v4();
        format!(
            "Task:StreamForwarder<{}>({})",
            std::any::type_name::<S::Item>(),
            uuid
        )
    }

    async fn interruptable_routine(mut self) -> Result<Self::Output, Error> {
        while let Some(item) = self.stream.next().await {
            let action = StreamItem::Item(item);
            self.recipient.act(action).await?;
        }
        let action = StreamItem::Done;
        self.recipient.act(action).await?;
        Ok(())
    }
}

pub(crate) struct AttachStream<S, M> {
    stream: S,
    tag: M,
}

impl<S, M> AttachStream<S, M> {
    pub fn new(stream: S, tag: M) -> Self {
        Self { stream, tag }
    }
}

impl<S, M> InstantAction for AttachStream<S, M>
where
    S: Stream + Send + 'static,
    M: Tag,
{
}

#[async_trait]
impl<T, S, M> InstantActionHandler<AttachStream<S, M>> for T
where
    T: Consumer<S::Item> + StreamAcceptor<S::Item>,
    S: Stream + Unpin + Send + 'static,
    S::Item: Send,
    M: Tag,
{
    async fn handle(
        &mut self,
        msg: AttachStream<S, M>,
        ctx: &mut Context<Self>,
    ) -> Result<(), Error> {
        let stream = msg.stream;
        let group = self.stream_group();
        ctx.attach(stream, msg.tag, group);
        Ok(())
    }
}