simple_hyper_client/connector/
mod.rs

1/* Copyright (c) Fortanix, Inc.
2 *
3 * This Source Code Form is subject to the terms of the Mozilla Public
4 * License, v. 2.0. If a copy of the MPL was not distributed with this
5 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
6
7use hyper::service::Service;
8use hyper::Uri;
9use hyper_util::client::legacy::connect::{Connected, Connection};
10use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
11
12use std::error::Error as StdError;
13use std::future::Future;
14use std::io;
15use std::pin::Pin;
16use std::sync::Arc;
17use std::task::{Context, Poll};
18
19pub mod http;
20pub mod hyper_adapter;
21
22pub use self::http::{ConnectError, HttpConnection, HttpConnector};
23pub use self::hyper_adapter::HyperConnectorAdapter;
24
25trait NetworkStream: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static {}
26
27impl<T> NetworkStream for T where T: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static {}
28
29/// A boxed network connection
30pub struct NetworkConnection(Box<dyn NetworkStream>);
31
32impl NetworkConnection {
33    pub fn new<S>(stream: S) -> Self
34    where
35        S: AsyncRead + AsyncWrite + Connection + Unpin + Send + 'static,
36    {
37        NetworkConnection(Box::new(stream))
38    }
39}
40
41impl Connection for NetworkConnection {
42    fn connected(&self) -> Connected {
43        self.0.connected()
44    }
45}
46
47impl AsyncRead for NetworkConnection {
48    fn poll_read(
49        self: Pin<&mut Self>,
50        cx: &mut Context<'_>,
51        buf: &mut ReadBuf<'_>,
52    ) -> Poll<io::Result<()>> {
53        Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
54    }
55}
56
57impl AsyncWrite for NetworkConnection {
58    fn poll_write(
59        self: Pin<&mut Self>,
60        cx: &mut Context<'_>,
61        buf: &[u8],
62    ) -> Poll<io::Result<usize>> {
63        Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
64    }
65
66    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
67        Pin::new(&mut self.get_mut().0).poll_flush(cx)
68    }
69
70    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
71        Pin::new(&mut self.get_mut().0).poll_shutdown(cx)
72    }
73}
74
75impl hyper::rt::Read for NetworkConnection {
76    fn poll_read(
77        self: Pin<&mut Self>,
78        cx: &mut Context<'_>,
79        mut buf: hyper::rt::ReadBufCursor<'_>,
80    ) -> Poll<Result<(), std::io::Error>> {
81        // SAFETY: Never uninitialize any bytes that may have been initialized before.
82        let mut tmp_buf = unsafe { ReadBuf::uninit(buf.as_mut()) };
83
84        let bytes_read = match AsyncRead::poll_read(self, cx, &mut tmp_buf) {
85            Poll::Ready(Ok(())) => tmp_buf.filled().len(),
86            other => return other,
87        };
88
89        // SAFETY: Advance by exactly the number of bytes we've just initialized.
90        unsafe { buf.advance(bytes_read) };
91
92        Poll::Ready(Ok(()))
93    }
94}
95
96impl hyper::rt::Write for NetworkConnection {
97    fn poll_write(
98        self: Pin<&mut Self>,
99        cx: &mut Context<'_>,
100        buf: &[u8],
101    ) -> Poll<Result<usize, std::io::Error>> {
102        AsyncWrite::poll_write(self, cx, buf)
103    }
104
105    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
106        AsyncWrite::poll_flush(self, cx)
107    }
108
109    fn poll_shutdown(
110        self: Pin<&mut Self>,
111        cx: &mut Context<'_>,
112    ) -> Poll<Result<(), std::io::Error>> {
113        AsyncWrite::poll_shutdown(self, cx)
114    }
115}
116
117/// Network connector trait with type erasure
118pub trait NetworkConnector: Send + Sync + 'static {
119    fn connect(
120        &self,
121        uri: Uri,
122    ) -> Pin<
123        Box<dyn Future<Output = Result<NetworkConnection, Box<dyn StdError + Send + Sync>>> + Send>,
124    >;
125}
126
127#[derive(Clone)]
128pub(crate) struct ConnectorAdapter(Arc<dyn NetworkConnector>);
129
130impl ConnectorAdapter {
131    pub fn new<T: NetworkConnector>(connector: T) -> Self {
132        Self(Arc::new(connector))
133    }
134}
135
136impl Service<Uri> for ConnectorAdapter {
137    type Response = NetworkConnection;
138    type Error = Box<dyn StdError + Send + Sync>;
139    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
140
141    fn call(&self, uri: Uri) -> Self::Future {
142        self.0.connect(uri)
143    }
144}
145
146impl tower_service::Service<Uri> for ConnectorAdapter {
147    type Response = NetworkConnection;
148    type Error = Box<dyn StdError + Send + Sync>;
149    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
150
151    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
152        Poll::Ready(Ok(()))
153    }
154
155    fn call(&mut self, uri: Uri) -> Self::Future {
156        self.0.connect(uri)
157    }
158}
simple_hyper_client/connector/mod.rs

simple_hyper_client/connector/
mod.rs
