http: rename EitherStream to MaybeTlsStream

And rename the enum values. Added an additional enum called Proxied.

The enum in now more specialized, but we only use it for the http client anyways.

src/tools/async_io.rs

@@ -1,4 +1,4 @@
-//! Generic AsyncRead/AsyncWrite utilities.
+//! AsyncRead/AsyncWrite utilities.
 
 use std::io;
 use std::os::unix::io::{AsRawFd, RawFd};
@@ -9,41 +9,52 @@ use futures::stream::{Stream, TryStream};
 use futures::ready;
 use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
 use tokio::net::TcpListener;
-use hyper::client::connect::Connection;
+use tokio_openssl::SslStream;
+use hyper::client::connect::{Connection, Connected};
 
-pub enum EitherStream<L, R> {
+/// Asynchronous stream, possibly encrypted and proxied
-    Left(L),
+///
-    Right(R),
+/// Usefule for HTTP client implementations using hyper.
+pub enum MaybeTlsStream<S> {
+    Normal(S),
+    Proxied(S),
+    Secured(SslStream<S>),
 }
 
-impl<L: AsyncRead + Unpin, R: AsyncRead + Unpin> AsyncRead for EitherStream<L, R> {
+impl<S: AsyncRead + AsyncWrite + Unpin> AsyncRead for MaybeTlsStream<S> {
     fn poll_read(
         self: Pin<&mut Self>,
         cx: &mut Context,
         buf: &mut ReadBuf,
     ) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
                 Pin::new(s).poll_read(cx, buf)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_read(cx, buf)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_read(cx, buf)
             }
         }
     }
 }
 
-impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L, R> {
+impl<S: AsyncRead + AsyncWrite + Unpin> AsyncWrite for MaybeTlsStream<S> {
     fn poll_write(
         self: Pin<&mut Self>,
         cx: &mut Context,
         buf: &[u8],
     ) -> Poll<Result<usize, io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
                 Pin::new(s).poll_write(cx, buf)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_write(cx, buf)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_write(cx, buf)
             }
         }
@@ -51,10 +62,13 @@ impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
                 Pin::new(s).poll_flush(cx)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_flush(cx)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_flush(cx)
             }
         }
@@ -62,25 +76,27 @@ impl<L: AsyncWrite + Unpin, R: AsyncWrite + Unpin> AsyncWrite for EitherStream<L
 
     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
         match self.get_mut() {
-            EitherStream::Left(ref mut s) => {
+            MaybeTlsStream::Normal(ref mut s) => {
                 Pin::new(s).poll_shutdown(cx)
             }
-            EitherStream::Right(ref mut s) => {
+            MaybeTlsStream::Proxied(ref mut s) => {
+                Pin::new(s).poll_shutdown(cx)
+            }
+            MaybeTlsStream::Secured(ref mut s) => {
                 Pin::new(s).poll_shutdown(cx)
             }
         }
     }
 }
 
-// we need this for crate::client::http_client:
+// we need this for the hyper http client
-impl Connection for EitherStream<
+impl <S: Connection + AsyncRead + AsyncWrite + Unpin> Connection for MaybeTlsStream<S>
-    tokio::net::TcpStream,
+{
-    Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>,
+    fn connected(&self) -> Connected {
-> {
-    fn connected(&self) -> hyper::client::connect::Connected {
         match self {
-            EitherStream::Left(s) => s.connected(),
+            MaybeTlsStream::Normal(s) => s.connected(),
-            EitherStream::Right(s) => s.get_ref().connected(),
+            MaybeTlsStream::Proxied(s) => s.connected().proxy(true),
+            MaybeTlsStream::Secured(s) => s.get_ref().connected(),
         }
     }
 }

src/tools/http.rs

@@ -10,9 +10,11 @@ use hyper::client::{Client, HttpConnector};
 use http::{Request, Response};
 use openssl::ssl::{SslConnector, SslMethod};
 use futures::*;
+use tokio::net::TcpStream;
+use tokio_openssl::SslStream;
 
 use crate::tools::{
-    async_io::EitherStream,
+    async_io::MaybeTlsStream,
     socket::{
         set_tcp_keepalive,
         PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
@@ -100,13 +102,8 @@ impl HttpsConnector {
     }
 }
 
-type MaybeTlsStream = EitherStream<
-    tokio::net::TcpStream,
-    Pin<Box<tokio_openssl::SslStream<tokio::net::TcpStream>>>,
->;
-
 impl hyper::service::Service<Uri> for HttpsConnector {
-    type Response = MaybeTlsStream;
+    type Response = MaybeTlsStream<TcpStream>;
     type Error = Error;
     #[allow(clippy::type_complexity)]
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
@@ -140,12 +137,11 @@ impl hyper::service::Service<Uri> for HttpsConnector {
             let _ = set_tcp_keepalive(conn.as_raw_fd(), PROXMOX_BACKUP_TCP_KEEPALIVE_TIME);
 
             if is_https {
-                let conn: tokio_openssl::SslStream<tokio::net::TcpStream> = tokio_openssl::SslStream::new(config?.into_ssl(&host)?, conn)?;
+                let mut conn: SslStream<TcpStream> = SslStream::new(config?.into_ssl(&host)?, conn)?;
-                let mut conn = Box::pin(conn);
+                Pin::new(&mut conn).connect().await?;
-                conn.as_mut().connect().await?;
+                Ok(MaybeTlsStream::Secured(conn))
-                Ok(MaybeTlsStream::Right(conn))
             } else {
-                Ok(MaybeTlsStream::Left(conn))
+                Ok(MaybeTlsStream::Normal(conn))
             }
         }.boxed()
     }