boomnet 0.0.78

Framework for building low latency clients on top of TCP.
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

//! Stream that can be used together with `MioSelector`.

use crate::service::select::Selectable;
use crate::stream::{ConnectionInfo, ConnectionInfoProvider};
use mio::event::Source;
use mio::net::TcpStream;
use mio::{Interest, Registry, Token};
use std::io::ErrorKind::{Interrupted, NotConnected, WouldBlock};
use std::io::{Read, Write};
use std::os::fd::{AsRawFd, RawFd};
use std::{io, net};

#[derive(Debug)]
pub struct MioStream {
    inner: TcpStream,
    connection_info: ConnectionInfo,
    connected: bool,
    can_read: bool,
    can_write: bool,
    buffer: Vec<u8>,
}

impl MioStream {
    fn new(inner: TcpStream, connection_info: ConnectionInfo) -> MioStream {
        Self {
            inner,
            connection_info,
            connected: false,
            can_read: false,
            can_write: false,
            buffer: Vec::with_capacity(4096),
        }
    }
}

impl AsRawFd for MioStream {
    fn as_raw_fd(&self) -> RawFd {
        self.inner.as_raw_fd()
    }
}

impl Selectable for MioStream {
    fn connected(&mut self) -> io::Result<bool> {
        if self.connected {
            return Ok(true);
        }
        match self.inner.peer_addr() {
            Ok(_) => {
                self.connected = true;
                // bypassing `can_write` as we can get to this state
                // only if the socket is writable
                self.inner.write_all(&self.buffer)?;
                self.buffer.clear();
                Ok(true)
            }
            Err(err) if err.kind() == NotConnected => Ok(false),
            Err(err) if err.kind() == Interrupted => Ok(false),
            Err(err) => Err(err),
        }
    }

    fn make_writable(&mut self) -> io::Result<()> {
        self.can_write = true;
        Ok(())
    }

    fn make_readable(&mut self) -> io::Result<()> {
        self.can_read = true;
        Ok(())
    }
}

impl Source for MioStream {
    fn register(&mut self, registry: &Registry, token: Token, interests: Interest) -> io::Result<()> {
        registry.register(&mut self.inner, token, interests)
    }

    fn reregister(&mut self, registry: &Registry, token: Token, interests: Interest) -> io::Result<()> {
        registry.reregister(&mut self.inner, token, interests)
    }

    fn deregister(&mut self, registry: &Registry) -> io::Result<()> {
        registry.deregister(&mut self.inner)
    }
}

impl Read for MioStream {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        if self.can_read {
            let read = self.inner.read(buf)?;
            if read < buf.len() {
                self.can_read = false;
            }
            return Ok(read);
        }
        Err(io::Error::from(WouldBlock))
    }
}

impl Write for MioStream {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        if !self.can_write {
            self.buffer.extend_from_slice(buf);
            return Ok(buf.len());
        }
        self.inner.write(buf)
    }

    fn flush(&mut self) -> io::Result<()> {
        self.inner.flush()
    }
}

impl ConnectionInfoProvider for MioStream {
    fn connection_info(&self) -> &ConnectionInfo {
        &self.connection_info
    }
}

pub trait IntoMioStream {
    fn into_mio_stream(self) -> MioStream;
}

impl<T> IntoMioStream for T
where
    T: Into<net::TcpStream>,
    T: ConnectionInfoProvider,
{
    fn into_mio_stream(self) -> MioStream {
        let connection_info = self.connection_info().clone();
        MioStream::new(TcpStream::from_std(self.into()), connection_info)
    }
}