noxious 1.0.4

The core library for noxious, a Rust port of Toxiproxy
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

use crate::signal::Stop;
use crate::socket::{ReadStream, WriteStream};
use bytes::{Bytes, BytesMut};
use futures::{Sink, SinkExt, Stream, StreamExt};
use std::io;
use std::pin::Pin;

use tokio_util::codec::{BytesCodec, FramedRead, FramedWrite};

pub(crate) type Read = FramedRead<ReadStream, BytesCodec>;
pub(crate) type Write = FramedWrite<WriteStream, BytesCodec>;

pub(crate) async fn forward(
    reader: &mut Read,
    writer: &mut Write,
    stop: &mut Stop,
) -> io::Result<()> {
    while !stop.stop_received() {
        let maybe_res: Option<io::Result<BytesMut>> = tokio::select! {
            res = reader.next() => res,
            _ = stop.recv() => None
        };
        if let Some(res) = maybe_res {
            match res {
                Ok(chunk) => {
                    if let Err(_err) = writer.send(chunk.into()).await {
                        // writer closed
                        break;
                    }
                }
                Err(err) => {
                    // reader closed
                    return Err(err);
                }
            }
        } else {
            // stop signal received
            break;
        }
    }
    Ok(())
}

pub(crate) async fn forward_read(
    mut reader: Read,
    mut writer: Pin<&mut impl Sink<Bytes>>,
    stop: &mut Stop,
) -> io::Result<Read> {
    while !stop.stop_received() {
        let maybe_res: Option<io::Result<BytesMut>> = tokio::select! {
            res = reader.next() => res,
            _ = stop.recv() => None
        };
        if let Some(res) = maybe_res {
            match res {
                Ok(chunk) => {
                    if let Err(_err) = writer.send(chunk.into()).await {
                        // writer channel closed
                        break;
                    }
                }
                Err(err) => {
                    // reader i/o error
                    return Err(err);
                }
            }
        } else {
            // stop signal received
            break;
        }
    }
    Ok(reader)
}

pub(crate) async fn forward_write(
    mut reader: Pin<&mut impl Stream<Item = Bytes>>,
    mut writer: Write,
    stop: &mut Stop,
) -> io::Result<Write> {
    while !stop.stop_received() {
        let maybe_chunk = tokio::select! {
            res = reader.next() => res,
            _ = stop.recv() => None
        };
        if let Some(chunk) = maybe_chunk {
            if writer.send(chunk).await.is_err() {
                // writer channel closed
                break;
            }
        } else {
            break;
        }
    }
    Ok(writer)
}