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
25pub trait NetworkStream:
26    AsyncRead + AsyncWrite + Connection + Unpin + Send + Sync + 'static
27{
28}
29
30impl<T> NetworkStream for T where
31    T: AsyncRead + AsyncWrite + Connection + Unpin + Send + Sync + 'static
32{
33}
34
35/// A boxed network connection
36pub struct NetworkConnection(Box<dyn NetworkStream>);
37
38impl NetworkConnection {
39    pub fn new<S>(stream: S) -> Self
40    where
41        S: AsyncRead + AsyncWrite + Connection + Unpin + Send + Sync + 'static,
42    {
43        NetworkConnection(Box::new(stream))
44    }
45}
46
47impl Connection for NetworkConnection {
48    fn connected(&self) -> Connected {
49        self.0.connected()
50    }
51}
52
53impl AsyncRead for NetworkConnection {
54    fn poll_read(
55        self: Pin<&mut Self>,
56        cx: &mut Context<'_>,
57        buf: &mut ReadBuf<'_>,
58    ) -> Poll<io::Result<()>> {
59        Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
60    }
61}
62
63impl AsyncWrite for NetworkConnection {
64    fn poll_write(
65        self: Pin<&mut Self>,
66        cx: &mut Context<'_>,
67        buf: &[u8],
68    ) -> Poll<io::Result<usize>> {
69        Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
70    }
71
72    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
73        Pin::new(&mut self.get_mut().0).poll_flush(cx)
74    }
75
76    fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
77        Pin::new(&mut self.get_mut().0).poll_shutdown(cx)
78    }
79}
80
81impl hyper::rt::Read for NetworkConnection {
82    fn poll_read(
83        self: Pin<&mut Self>,
84        cx: &mut Context<'_>,
85        mut buf: hyper::rt::ReadBufCursor<'_>,
86    ) -> Poll<Result<(), std::io::Error>> {
87        // SAFETY: Never uninitialize any bytes that may have been initialized before.
88        let mut tmp_buf = unsafe { ReadBuf::uninit(buf.as_mut()) };
89
90        let bytes_read = match AsyncRead::poll_read(self, cx, &mut tmp_buf) {
91            Poll::Ready(Ok(())) => tmp_buf.filled().len(),
92            other => return other,
93        };
94
95        // SAFETY: Advance by exactly the number of bytes we've just initialized.
96        unsafe { buf.advance(bytes_read) };
97
98        Poll::Ready(Ok(()))
99    }
100}
101
102impl hyper::rt::Write for NetworkConnection {
103    fn poll_write(
104        self: Pin<&mut Self>,
105        cx: &mut Context<'_>,
106        buf: &[u8],
107    ) -> Poll<Result<usize, std::io::Error>> {
108        AsyncWrite::poll_write(self, cx, buf)
109    }
110
111    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
112        AsyncWrite::poll_flush(self, cx)
113    }
114
115    fn poll_shutdown(
116        self: Pin<&mut Self>,
117        cx: &mut Context<'_>,
118    ) -> Poll<Result<(), std::io::Error>> {
119        AsyncWrite::poll_shutdown(self, cx)
120    }
121}
122
123/// Network connector trait with type erasure
124pub trait NetworkConnector: Send + Sync + 'static {
125    fn connect(
126        &self,
127        uri: Uri,
128    ) -> Pin<
129        Box<dyn Future<Output = Result<NetworkConnection, Box<dyn StdError + Send + Sync>>> + Send>,
130    >;
131}
132
133#[derive(Clone)]
134pub(crate) struct ConnectorAdapter(Arc<dyn NetworkConnector>);
135
136impl ConnectorAdapter {
137    pub fn new<T: NetworkConnector>(connector: T) -> Self {
138        Self(Arc::new(connector))
139    }
140}
141
142impl Service<Uri> for ConnectorAdapter {
143    type Response = NetworkConnection;
144    type Error = Box<dyn StdError + Send + Sync>;
145    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
146
147    fn call(&self, uri: Uri) -> Self::Future {
148        self.0.connect(uri)
149    }
150}
151
152impl tower_service::Service<Uri> for ConnectorAdapter {
153    type Response = NetworkConnection;
154    type Error = Box<dyn StdError + Send + Sync>;
155    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
156
157    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
158        Poll::Ready(Ok(()))
159    }
160
161    fn call(&mut self, uri: Uri) -> Self::Future {
162        self.0.connect(uri)
163    }
164}
simple_hyper_client/connector/mod.rs

simple_hyper_client/connector/
mod.rs
