coerce 0.8.6

Async actor runtime and distributed systems framework
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

use crate::remote::system::RemoteActorSystem;

use std::future::Future;
use std::io::Error;

use bytes::BytesMut;
use std::pin::Pin;
use std::task::{Context, Poll};

use futures::StreamExt;
use tokio_util::codec::{FramedRead, LengthDelimitedCodec};

pub mod client;
pub mod message;
pub mod metrics;
pub mod proto;
pub mod security;
pub mod server;

pub trait StreamData: 'static + Send + Sync + Sized {
    fn read_from_bytes(data: Vec<u8>) -> Option<Self>;

    fn write_to_bytes(&self) -> Option<Vec<u8>>;
}

#[async_trait]
pub trait StreamReceiver {
    type Message: StreamData;

    async fn on_receive(&mut self, msg: Self::Message, sys: &RemoteActorSystem);

    async fn on_close(&mut self, sys: &RemoteActorSystem);

    fn on_deserialisation_failed(&mut self);

    fn on_stream_lost(&mut self, error: Error);

    async fn close(&mut self);

    fn should_close(&self) -> bool;
}

pub struct StreamReceiverFuture<S: tokio::io::AsyncRead> {
    stream: FramedRead<S, LengthDelimitedCodec>,
    stop_rx: tokio::sync::oneshot::Receiver<bool>,
}

impl<S: tokio::io::AsyncRead> StreamReceiverFuture<S> {
    pub fn new(
        stream: FramedRead<S, LengthDelimitedCodec>,
        stop_rx: tokio::sync::oneshot::Receiver<bool>,
    ) -> StreamReceiverFuture<S> {
        StreamReceiverFuture { stream, stop_rx }
    }
}

impl<S: tokio::io::AsyncRead> tokio_stream::Stream for StreamReceiverFuture<S>
where
    S: Unpin,
{
    type Item = Option<BytesMut>;

    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Option<BytesMut>>> {
        if let Poll::Ready(Ok(true)) = Pin::new(&mut self.stop_rx).poll(cx) {
            return Poll::Ready(None);
        }

        let result: Option<Result<BytesMut, Error>> =
            futures::ready!(Pin::new(&mut self.stream).poll_next(cx));

        Poll::Ready(match result {
            Some(Ok(message)) => Some(Some(message)),
            Some(Err(e)) => {
                error!("{:?}", e);
                Some(None)
            }
            None => None,
        })
    }
}

pub async fn receive_loop<R: StreamReceiver, S: tokio::io::AsyncRead + Unpin>(
    mut system: RemoteActorSystem,
    read: FramedRead<S, LengthDelimitedCodec>,
    mut receiver: R,
) where
    R: Send,
{
    let mut reader = read;
    while let Some(res) = reader.next().await {
        match res {
            Ok(res) => match R::Message::read_from_bytes(res.to_vec()) {
                Some(msg) => {
                    receiver.on_receive(msg, &system).await;
                    if receiver.should_close() {
                        break;
                    }
                }
                None => {
                    // TODO: either pass the buffer into here or more context, this is pretty useless at the moment..
                    receiver.on_deserialisation_failed();
                }
            },
            Err(e) => {
                receiver.on_stream_lost(e);
                break;
            }
        }
    }

    receiver.on_close(&mut system).await;
}