-
Notifications
You must be signed in to change notification settings - Fork 49
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #59 from Fishrock123/async-h1-pooling
feat: h1 connection pooling
- Loading branch information
Showing
4 changed files
with
256 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| use std::fmt::Debug; | ||
| use std::net::SocketAddr; | ||
| use std::pin::Pin; | ||
|
|
||
| use async_std::net::TcpStream; | ||
| use async_trait::async_trait; | ||
| use deadpool::managed::{Manager, Object, RecycleResult}; | ||
| use futures::io::{AsyncRead, AsyncWrite}; | ||
| use futures::task::{Context, Poll}; | ||
|
|
||
| #[derive(Clone, Debug)] | ||
| pub(crate) struct TcpConnection { | ||
| addr: SocketAddr, | ||
| } | ||
| impl TcpConnection { | ||
| pub(crate) fn new(addr: SocketAddr) -> Self { | ||
| Self { addr } | ||
| } | ||
| } | ||
|
|
||
| pub(crate) struct TcpConnWrapper { | ||
| conn: Object<TcpStream, std::io::Error>, | ||
| } | ||
| impl TcpConnWrapper { | ||
| pub(crate) fn new(conn: Object<TcpStream, std::io::Error>) -> Self { | ||
| Self { conn } | ||
| } | ||
| } | ||
|
|
||
| impl AsyncRead for TcpConnWrapper { | ||
| fn poll_read( | ||
| mut self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| buf: &mut [u8], | ||
| ) -> Poll<Result<usize, std::io::Error>> { | ||
| Pin::new(&mut *self.conn).poll_read(cx, buf) | ||
| } | ||
| } | ||
|
|
||
| impl AsyncWrite for TcpConnWrapper { | ||
| fn poll_write( | ||
| mut self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| buf: &[u8], | ||
| ) -> Poll<std::io::Result<usize>> { | ||
| Pin::new(&mut *self.conn).poll_write(cx, buf) | ||
| } | ||
|
|
||
| fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> { | ||
| Pin::new(&mut *self.conn).poll_flush(cx) | ||
| } | ||
|
|
||
| fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> { | ||
| Pin::new(&mut *self.conn).poll_close(cx) | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl Manager<TcpStream, std::io::Error> for TcpConnection { | ||
| async fn create(&self) -> Result<TcpStream, std::io::Error> { | ||
| Ok(TcpStream::connect(self.addr).await?) | ||
| } | ||
|
|
||
| async fn recycle(&self, _conn: &mut TcpStream) -> RecycleResult<std::io::Error> { | ||
| Ok(()) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,91 @@ | ||
| use std::fmt::Debug; | ||
| use std::net::SocketAddr; | ||
| use std::pin::Pin; | ||
|
|
||
| use async_std::net::TcpStream; | ||
| use async_trait::async_trait; | ||
| use deadpool::managed::{Manager, Object, RecycleResult}; | ||
| use futures::io::{AsyncRead, AsyncWrite}; | ||
| use futures::task::{Context, Poll}; | ||
|
|
||
| #[cfg(not(feature = "h1_client_rustls"))] | ||
| use async_native_tls::TlsStream; | ||
| #[cfg(feature = "h1_client_rustls")] | ||
| use async_tls::client::TlsStream; | ||
|
|
||
| use crate::Error; | ||
|
|
||
| #[derive(Clone, Debug)] | ||
| pub(crate) struct TlsConnection { | ||
| host: String, | ||
| addr: SocketAddr, | ||
| } | ||
| impl TlsConnection { | ||
| pub(crate) fn new(host: String, addr: SocketAddr) -> Self { | ||
| Self { host, addr } | ||
| } | ||
| } | ||
|
|
||
| pub(crate) struct TlsConnWrapper { | ||
| conn: Object<TlsStream<TcpStream>, Error>, | ||
| } | ||
| impl TlsConnWrapper { | ||
| pub(crate) fn new(conn: Object<TlsStream<TcpStream>, Error>) -> Self { | ||
| Self { conn } | ||
| } | ||
| } | ||
|
|
||
| impl AsyncRead for TlsConnWrapper { | ||
| fn poll_read( | ||
| mut self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| buf: &mut [u8], | ||
| ) -> Poll<Result<usize, std::io::Error>> { | ||
| Pin::new(&mut *self.conn).poll_read(cx, buf) | ||
| } | ||
| } | ||
|
|
||
| impl AsyncWrite for TlsConnWrapper { | ||
| fn poll_write( | ||
| mut self: Pin<&mut Self>, | ||
| cx: &mut Context<'_>, | ||
| buf: &[u8], | ||
| ) -> Poll<std::io::Result<usize>> { | ||
| Pin::new(&mut *self.conn).poll_write(cx, buf) | ||
| } | ||
|
|
||
| fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> { | ||
| Pin::new(&mut *self.conn).poll_flush(cx) | ||
| } | ||
|
|
||
| fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> { | ||
| Pin::new(&mut *self.conn).poll_close(cx) | ||
| } | ||
| } | ||
|
|
||
| #[async_trait] | ||
| impl Manager<TlsStream<TcpStream>, Error> for TlsConnection { | ||
| async fn create(&self) -> Result<TlsStream<TcpStream>, Error> { | ||
| let raw_stream = async_std::net::TcpStream::connect(self.addr).await?; | ||
| let tls_stream = add_tls(&self.host, raw_stream).await?; | ||
| Ok(tls_stream) | ||
| } | ||
|
|
||
| async fn recycle(&self, _conn: &mut TlsStream<TcpStream>) -> RecycleResult<Error> { | ||
| Ok(()) | ||
| } | ||
| } | ||
|
|
||
| #[cfg(not(feature = "h1_client_rustls"))] | ||
| async fn add_tls( | ||
| host: &str, | ||
| stream: TcpStream, | ||
| ) -> Result<async_native_tls::TlsStream<TcpStream>, async_native_tls::Error> { | ||
| async_native_tls::connect(host, stream).await | ||
| } | ||
|
|
||
| #[cfg(feature = "h1_client_rustls")] | ||
| async fn add_tls(host: &str, stream: TcpStream) -> Result<TlsStream<TcpStream>, std::io::Error> { | ||
| let connector = async_tls::TlsConnector::default(); | ||
| connector.connect(host, stream).await | ||
| } |