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
use std::{
pin::Pin,
task::{Context, Poll},
};
use futures::{
channel::{mpsc, mpsc::SendError},
prelude::*,
sink::{Sink, SinkErrInto},
StreamExt,
};
use pin_project::pin_project;
use crate::{Error, InEvent, OutEvent};
pub struct MpscHandleStream {
receiver: mpsc::Receiver<InEvent>,
}
impl MpscHandleStream {
pub fn new(receiver: mpsc::Receiver<InEvent>) -> MpscHandleStream {
MpscHandleStream { receiver }
}
}
impl Stream for MpscHandleStream {
type Item = InEvent;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
self.receiver.poll_next_unpin(cx)
}
}
#[pin_project]
pub struct MpscHandleSink {
#[pin]
sender: SinkErrInto<mpsc::Sender<OutEvent>, OutEvent, Error>,
}
impl MpscHandleSink {
pub fn new(sender: mpsc::Sender<OutEvent>) -> MpscHandleSink {
MpscHandleSink {
sender: sender.sink_err_into(),
}
}
}
impl Sink<OutEvent> for MpscHandleSink {
type Error = Error;
fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project().sender.poll_ready(cx)
}
fn start_send(self: Pin<&mut Self>, item: OutEvent) -> Result<(), Self::Error> {
self.project().sender.start_send(item)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project().sender.poll_flush(cx)
}
fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.project().sender.poll_close(cx)
}
}
impl From<SendError> for Error {
fn from(s: SendError) -> Self {
Error::Other(format!("Sink error: {}", s))
}
}