rammux 0.1.2

Asynchronous stream multiplexer
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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use std::time::Duration;

use bytes::Bytes;
use futures::{SinkExt, StreamExt};
use rammux::{
    config::{RammuxConfig, RammuxRole},
    connection::{RammuxConnection, RammuxProgress},
};
use tokio::io::DuplexStream;

const DATA: Bytes = Bytes::from_static(&[37; 64 * 1024]);

/// This example presents the flow control mechanism implemented in rammux.
///
/// See [PROTOCOL.md](https://github.com/Razz4780/rammux/blob/main/PROTOCOL.md) for more info.
#[tokio::main]
async fn main() {
    let (io_1, io_2) = tokio::io::duplex(4096);
    let client = RammuxConnection::new(RammuxRole::Client, io_1, RammuxConfig::new());
    let server = RammuxConnection::new(RammuxRole::Server, io_2, RammuxConfig::new());
    tokio::join!(run_client(client), run_server(server));
}

/// Opens two streams and floods both with repeated [`DATA`].
async fn run_client(mut rammux: RammuxConnection<DuplexStream>) {
    let role = rammux.role();

    for _ in 0..2 {
        let mut stream = rammux
            .try_start_outbound()
            .expect("connection should not fail")
            .expect("config allows for at least two outbound streams");
        println!("[{role}] Started stream {}", stream.id());
        tokio::spawn(async move {
            loop {
                if stream.send(DATA.clone()).await.is_ok() {
                    println!(
                        "[{role}] Sent and flushed {} bytes in stream {}",
                        DATA.len(),
                        stream.id(),
                    );
                } else {
                    println!("[{role}] Stream {} was closed.", stream.id());
                    break;
                }
            }
        });
    }

    let downgraded = loop {
        let progress = rammux.progress().await.unwrap();
        match progress {
            RammuxProgress::Inbound(..) => panic!("we don't expect any more streams"),
            RammuxProgress::Downgraded(downgraded) => {
                println!("[{role}] Peer started downgrade.");
                break downgraded;
            },
            RammuxProgress::Empty => {},
        }
    };

    downgraded
        .with_shutdown()
        .await
        .expect("connection should not fail");
    println!("[{role}] Downgrade finished.");
}

/// Accepts two streams. Does not read from the first stream, but reads slowly from the second stream.
/// Second stream remains functional even though the first stream is blocked.
///
/// Downgrades the connection 10 seconds after accepting both streams.
async fn run_server(mut rammux: RammuxConnection<DuplexStream>) {
    let role = rammux.role();

    let mut missing_streams = 2_usize;

    loop {
        let progress = rammux.progress().await.expect("connection should not fail");
        match progress {
            RammuxProgress::Downgraded(..) => {
                panic!("we don't expect downgrade from the other side")
            },
            RammuxProgress::Inbound(mut stream) => {
                missing_streams = missing_streams
                    .checked_sub(1)
                    .expect("we don't expect more than two streams");
                if missing_streams == 0 {
                    tokio::spawn(async move {
                        println!(
                            "[{role}] Accepted new stream {}, reading data slowly.",
                            stream.id()
                        );
                        while let Some(chunk) = stream.next().await {
                            println!(
                                "[{role}] Read {} bytes from stream {}.",
                                chunk.len(),
                                stream.id()
                            );
                            tokio::time::sleep(Duration::from_secs(1)).await;
                        }
                    });
                    break;
                } else {
                    tokio::spawn(async move {
                        println!(
                            "[{role}] Accepted new stream {}, not reading the data.",
                            stream.id()
                        );
                        let _ = stream;
                    });
                }
            },
            RammuxProgress::Empty => {},
        }
    }

    let mut sleep = Box::pin(tokio::time::sleep(Duration::from_secs(10)));
    loop {
        tokio::select! {
            _ = &mut sleep => break,
            progress = rammux.progress() => {
                let progress = progress.expect("connection should not fail");
                match progress {
                    RammuxProgress::Inbound(..) => panic!("we don't expect any more streams"),
                    RammuxProgress::Downgraded(..) => panic!("we don't expect downgrade from the other side"),
                    RammuxProgress::Empty => {},
                }
            }
        }
    }

    println!("[{role}] Starting downgrade.");
    rammux
        .downgrade()
        .expect("was not downgraded")
        .with_shutdown()
        .await
        .expect("connection should not fail");
    println!("[{role}] Downgrade finished.");
}