sl-messages 1.1.0

Message exchange for MPC protocols
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
113

// Copyright (c) Silence Laboratories Pte. Ltd. All Rights Reserved.
// This software is licensed under the Silence Laboratories License Agreement.

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

use tokio::sync::mpsc::{channel, Receiver};
use tokio_util::sync::PollSender;

use futures_util::{Sink, Stream};

use super::MessageSendError;

/// Simple two party message relay
pub struct SimplePair {
    out: PollSender<Vec<u8>>,
    inq: Receiver<Vec<u8>>,
}

impl SimplePair {
    /// Create a connected pair of message relays
    pub fn connect() -> (SimplePair, SimplePair) {
        let (out_tx, out_rx) = channel(1);
        let (in_tx, in_rx) = channel(1);

        let client = SimplePair {
            out: PollSender::new(out_tx),
            inq: in_rx,
        };

        let server = SimplePair {
            out: PollSender::new(in_tx),
            inq: out_rx,
        };

        (client, server)
    }
}

impl Stream for SimplePair {
    type Item = Vec<u8>;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        self.get_mut().inq.poll_recv(cx)
    }
}

impl Sink<Vec<u8>> for SimplePair {
    type Error = MessageSendError;

    fn poll_ready(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), Self::Error>> {
        self.get_mut()
            .out
            .poll_reserve(cx)
            .map_err(|_| MessageSendError)
    }

    fn start_send(
        self: Pin<&mut Self>,
        item: Vec<u8>,
    ) -> Result<(), Self::Error> {
        self.get_mut()
            .out
            .send_item(item)
            .map_err(|_| MessageSendError)
    }

    fn poll_flush(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
    ) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn poll_close(
        self: Pin<&mut Self>,
        _cx: &mut Context<'_>,
    ) -> Poll<Result<(), Self::Error>> {
        self.get_mut().out.close();

        Poll::Ready(Ok(()))
    }
}

#[cfg(test)]
mod test {
    use super::*;

    use futures_util::{SinkExt, StreamExt};

    #[tokio::test]
    async fn pair() {
        let (mut p1, mut p2) = SimplePair::connect();

        p1.send(vec![1, 2]).await.unwrap();

        let m = p2.next().await.unwrap();
        assert_eq!(&m, &[1, 2]);

        p2.send(vec![3, 4]).await.unwrap();

        let m = p1.next().await.unwrap();
        assert_eq!(&m, &[3, 4]);
    }
}