ruchei 0.1.2

Utilities for working with many streams
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

//! [`ruchei::multicast::bufferless`] with [`ruchei::echo::buffered`]

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

use async_net::TcpListener;
use futures_util::{Sink, SinkExt, Stream, StreamExt, TryStream, TryStreamExt, future::ready};
use pin_project::pin_project;
use ruchei::{
    concurrent::ConcurrentExt,
    connection_item::ConnectionItemExt,
    echo::bufferless::EchoBufferless,
    multicast::trie::{MulticastTrie, SubRequest},
    poll_on_wake::PollOnWakeExt,
};

#[pin_project]
struct AutoSub<S> {
    #[pin]
    inner: S,
    subbed: bool,
}

impl<S> AutoSub<S> {
    fn new(inner: S) -> Self {
        Self {
            inner,
            subbed: false,
        }
    }
}

impl<S: TryStream> Stream for AutoSub<S> {
    type Item = Result<SubRequest<&'static str, S::Ok>, S::Error>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();
        if !std::mem::replace(this.subbed, true) {
            Poll::Ready(Some(Ok(SubRequest::Sub("prefix"))))
        } else {
            this.inner.try_poll_next(cx).map_ok(SubRequest::Other)
        }
    }
}

impl<S: Sink<T>, T> Sink<T> for AutoSub<S> {
    type Error = S::Error;

    fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.project().inner.poll_ready(cx)
    }

    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
        self.project().inner.start_send(item)
    }

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.project().inner.poll_flush(cx)
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.project().inner.poll_close(cx)
    }
}

#[async_std::main]
async fn main() {
    TcpListener::bind("127.0.0.1:8080")
        .await
        .unwrap()
        .incoming()
        .poll_on_wake()
        .filter_map(|r| async { r.ok() })
        .map(async_tungstenite::accept_async)
        .concurrent()
        .filter_map(|r| async { r.ok() })
        .map(AutoSub::new)
        .map(|s| s.with(|(_, m)| ready(Ok::<_, async_tungstenite::tungstenite::Error>(m))))
        .multicast_trie()
        .connection_item_ignore()
        .map_ok(|m| ("prefix", m))
        .echo_bufferless()
        .await
        .unwrap();
}