[pbs-devel] [PATCH proxmox-backup 3/8] move MaybeTlsStream wrapper to proxmox_http
Fabian Grünbichler
f.gruenbichler at proxmox.com
Fri May 14 15:44:52 CEST 2021
Signed-off-by: Fabian Grünbichler <f.gruenbichler at proxmox.com>
---
Notes:
requires proxmox patch #6
Cargo.toml | 2 +-
src/tools/async_io.rs | 119 +-----------------------------------------
src/tools/http.rs | 6 +--
3 files changed, 4 insertions(+), 123 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index 997017c0..df649c1b 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -57,7 +57,7 @@ proxmox = { version = "0.11.4", features = [ "sortable-macro", "api-macro" ] }
#proxmox = { git = "git://git.proxmox.com/git/proxmox", version = "0.1.2", features = [ "sortable-macro", "api-macro" ] }
#proxmox = { path = "../proxmox/proxmox", features = [ "sortable-macro", "api-macro" ] }
proxmox-fuse = "0.1.1"
-proxmox-http = { version = "0.1.0", path = "../proxmox/proxmox-http", features = [ "websocket" ] }
+proxmox-http = { version = "0.1.0", path = "../proxmox/proxmox-http", features = [ "http-helpers", "websocket" ] }
pxar = { version = "0.10.1", features = [ "tokio-io" ] }
#pxar = { path = "../pxar", features = [ "tokio-io" ] }
regex = "1.2"
diff --git a/src/tools/async_io.rs b/src/tools/async_io.rs
index 83110912..66d38094 100644
--- a/src/tools/async_io.rs
+++ b/src/tools/async_io.rs
@@ -1,131 +1,14 @@
//! AsyncRead/AsyncWrite utilities.
-use std::io;
use std::os::unix::io::{AsRawFd, RawFd};
use std::pin::Pin;
use std::task::{Context, Poll};
use futures::stream::{Stream, TryStream};
use futures::ready;
-use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
+use tokio::io::{AsyncRead, AsyncWrite};
use tokio::net::TcpListener;
-use tokio_openssl::SslStream;
-use hyper::client::connect::{Connection, Connected};
-/// Asynchronous stream, possibly encrypted and proxied
-///
-/// Usefule for HTTP client implementations using hyper.
-pub enum MaybeTlsStream<S> {
- Normal(S),
- Proxied(S),
- Secured(SslStream<S>),
-}
-
-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() {
- MaybeTlsStream::Normal(ref mut s) => {
- Pin::new(s).poll_read(cx, buf)
- }
- 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<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() {
- MaybeTlsStream::Normal(ref mut s) => {
- Pin::new(s).poll_write(cx, buf)
- }
- MaybeTlsStream::Proxied(ref mut s) => {
- Pin::new(s).poll_write(cx, buf)
- }
- MaybeTlsStream::Secured(ref mut s) => {
- Pin::new(s).poll_write(cx, buf)
- }
- }
- }
-
- fn poll_write_vectored(
- self: Pin<&mut Self>,
- cx: &mut Context<'_>,
- bufs: &[io::IoSlice<'_>],
- ) -> Poll<Result<usize, io::Error>> {
- match self.get_mut() {
- MaybeTlsStream::Normal(ref mut s) => {
- Pin::new(s).poll_write_vectored(cx, bufs)
- }
- MaybeTlsStream::Proxied(ref mut s) => {
- Pin::new(s).poll_write_vectored(cx, bufs)
- }
- MaybeTlsStream::Secured(ref mut s) => {
- Pin::new(s).poll_write_vectored(cx, bufs)
- }
- }
- }
-
- fn is_write_vectored(&self) -> bool {
- match self {
- MaybeTlsStream::Normal(s) => s.is_write_vectored(),
- MaybeTlsStream::Proxied(s) => s.is_write_vectored(),
- MaybeTlsStream::Secured(s) => s.is_write_vectored(),
- }
- }
-
- fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
- match self.get_mut() {
- MaybeTlsStream::Normal(ref mut s) => {
- Pin::new(s).poll_flush(cx)
- }
- MaybeTlsStream::Proxied(ref mut s) => {
- Pin::new(s).poll_flush(cx)
- }
- MaybeTlsStream::Secured(ref mut s) => {
- Pin::new(s).poll_flush(cx)
- }
- }
- }
-
- fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Result<(), io::Error>> {
- match self.get_mut() {
- MaybeTlsStream::Normal(ref mut s) => {
- Pin::new(s).poll_shutdown(cx)
- }
- 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 the hyper http client
-impl <S: Connection + AsyncRead + AsyncWrite + Unpin> Connection for MaybeTlsStream<S>
-{
- fn connected(&self) -> Connected {
- match self {
- MaybeTlsStream::Normal(s) => s.connected(),
- MaybeTlsStream::Proxied(s) => s.connected().proxy(true),
- MaybeTlsStream::Secured(s) => s.get_ref().connected(),
- }
- }
-}
/// Tokio's `Incoming` now is a reference type and hyper's `AddrIncoming` misses some standard
/// stuff like `AsRawFd`, so here's something implementing hyper's `Accept` from a `TcpListener`
diff --git a/src/tools/http.rs b/src/tools/http.rs
index 1d96c70f..0f5b8470 100644
--- a/src/tools/http.rs
+++ b/src/tools/http.rs
@@ -19,11 +19,9 @@ use tokio::{
use tokio_openssl::SslStream;
use proxmox::sys::linux::socket::set_tcp_keepalive;
+use proxmox_http::http::MaybeTlsStream;
-use crate::tools::{
- PROXMOX_BACKUP_TCP_KEEPALIVE_TIME,
- async_io::MaybeTlsStream,
-};
+use crate::tools::PROXMOX_BACKUP_TCP_KEEPALIVE_TIME;
// Build a http::uri::Authority ("host:port"), use '[..]' around IPv6 addresses
pub(crate) fn build_authority(host: &str, port: u16) -> Result<Authority, Error> {
--
2.20.1
More information about the pbs-devel
mailing list