camber 0.1.1

Opinionated async Rust for IO-bound services on top of Tokio
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

use crate::RuntimeError;
use std::net::SocketAddr;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

/// Async TLS stream wrapping either a client or server `tokio_rustls` stream.
///
/// Provides the same read/write API as `TcpStream`, plus `peer_certificates()`
/// for TLS inspection (e.g. cert probing).
pub struct TlsStream {
    inner: TlsStreamInner,
}

enum TlsStreamInner {
    Client(tokio_rustls::client::TlsStream<tokio::net::TcpStream>),
    Server(tokio_rustls::server::TlsStream<tokio::net::TcpStream>),
}

macro_rules! dispatch {
    ($self:expr, |$s:ident| $body:expr) => {
        match &mut $self.inner {
            TlsStreamInner::Client($s) => $body,
            TlsStreamInner::Server($s) => $body,
        }
    };
    (ref $self:expr, |$s:ident| $body:expr) => {
        match &$self.inner {
            TlsStreamInner::Client($s) => $body,
            TlsStreamInner::Server($s) => $body,
        }
    };
}

impl std::fmt::Debug for TlsStream {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TlsStream")
            .field("peer_addr", &self.peer_addr().ok())
            .finish()
    }
}

impl TlsStream {
    pub(crate) fn from_client(
        stream: tokio_rustls::client::TlsStream<tokio::net::TcpStream>,
    ) -> Self {
        Self {
            inner: TlsStreamInner::Client(stream),
        }
    }

    pub(crate) fn from_server(
        stream: tokio_rustls::server::TlsStream<tokio::net::TcpStream>,
    ) -> Self {
        Self {
            inner: TlsStreamInner::Server(stream),
        }
    }

    /// Read data into the buffer, returning the number of bytes read.
    pub async fn read(&mut self, read_buf: &mut [u8]) -> Result<usize, RuntimeError> {
        let bytes_read = dispatch!(self, |s| s.read(read_buf).await?);
        Ok(bytes_read)
    }

    /// Write all bytes from the buffer.
    pub async fn write_all(&mut self, write_buf: &[u8]) -> Result<(), RuntimeError> {
        dispatch!(self, |s| s.write_all(write_buf).await?);
        Ok(())
    }

    /// Shut down the write half of the stream.
    pub async fn shutdown(&mut self) -> Result<(), RuntimeError> {
        dispatch!(self, |s| s.shutdown().await?);
        Ok(())
    }

    /// Returns the remote address this stream is connected to.
    pub fn peer_addr(&self) -> Result<SocketAddr, RuntimeError> {
        let addr = dispatch!(ref self, |s| s.get_ref().0).peer_addr()?;
        Ok(addr)
    }

    /// Returns the local address this stream is bound to.
    pub fn local_addr(&self) -> Result<SocketAddr, RuntimeError> {
        let addr = dispatch!(ref self, |s| s.get_ref().0).local_addr()?;
        Ok(addr)
    }

    /// Returns the peer's TLS certificates, if available.
    pub fn peer_certificates(&self) -> Option<&[rustls::pki_types::CertificateDer<'static>]> {
        dispatch!(ref self, |s| s.get_ref().1.peer_certificates())
    }
}

impl tokio::io::AsyncRead for TlsStream {
    fn poll_read(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &mut tokio::io::ReadBuf<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        match &mut self.inner {
            TlsStreamInner::Client(s) => std::pin::Pin::new(s).poll_read(cx, buf),
            TlsStreamInner::Server(s) => std::pin::Pin::new(s).poll_read(cx, buf),
        }
    }
}

impl tokio::io::AsyncWrite for TlsStream {
    fn poll_write(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
        buf: &[u8],
    ) -> std::task::Poll<std::io::Result<usize>> {
        match &mut self.inner {
            TlsStreamInner::Client(s) => std::pin::Pin::new(s).poll_write(cx, buf),
            TlsStreamInner::Server(s) => std::pin::Pin::new(s).poll_write(cx, buf),
        }
    }

    fn poll_flush(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        match &mut self.inner {
            TlsStreamInner::Client(s) => std::pin::Pin::new(s).poll_flush(cx),
            TlsStreamInner::Server(s) => std::pin::Pin::new(s).poll_flush(cx),
        }
    }

    fn poll_shutdown(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<std::io::Result<()>> {
        match &mut self.inner {
            TlsStreamInner::Client(s) => std::pin::Pin::new(s).poll_shutdown(cx),
            TlsStreamInner::Server(s) => std::pin::Pin::new(s).poll_shutdown(cx),
        }
    }
}