fix: remove unused try buffer

Author: conblem

src/api/mod.rs

@@ -1,6 +1,6 @@
 use anyhow::Result;
 use futures_util::future::OptionFuture;
-use futures_util::FutureExt;
+use futures_util::{FutureExt, TryStreamExt};
 use metrics::{metrics, metrics_wrapper};
 use sqlx::PgPool;
 use tokio::net::TcpListener;
@@ -28,12 +28,12 @@ pub async fn new(
     let routes = routes::routes(pool.clone());
 
     let http = http
-        .map(move |http| proxy::wrap(http, http_proxy))
+        .map(move |http| proxy::wrap(http, http_proxy).try_buffer_unordered(100))
         .map(|http| warp::serve(routes.clone()).serve_incoming(http))
         .map(tokio::spawn);
 
     let prom = prom
-        .map(move |prom| proxy::wrap(prom, prom_proxy))
+        .map(move |prom| proxy::wrap(prom, prom_proxy).try_buffer_unordered(100))
         .map(|prom| warp::serve(metrics()).serve_incoming(prom))
         .map(tokio::spawn);
 

src/api/proxy.rs

@@ -19,7 +19,7 @@ use crate::config::ProxyProtocol;
 pub(super) fn wrap(
     listener: TcpListener,
     proxy: ProxyProtocol,
-) -> impl Stream<Item = IoResult<ProxyStream>> + Send {
+) -> impl Stream<Item = IoResult<impl Future<Output = IoResult<ProxyStream>>>> + Send {
     TcpListenerStream::new(listener)
         .map_ok(move |stream| stream.source(proxy))
         .map_ok(|mut conn| {
@@ -40,7 +40,6 @@ pub(super) fn wrap(
             }
             .instrument(span)
         })
-        .try_buffer_unordered(100)
 }
 
 pub(super) trait ToProxyStream {

src/api/tls.rs

@@ -1,10 +1,11 @@
 use anyhow::{anyhow, Error, Result};
 use futures_util::stream::{repeat, Stream};
-use futures_util::{StreamExt, TryStreamExt};
+use futures_util::{StreamExt, TryStreamExt, TryFutureExt};
 use parking_lot::RwLock;
 use rustls::internal::pemfile::{certs, pkcs8_private_keys};
 use rustls::{NoClientAuth, ServerConfig};
 use sqlx::PgPool;
+use std::future::Future;
 use std::sync::Arc;
 use tokio::io::{AsyncRead, AsyncWrite, Result as IoResult};
 use tokio_rustls::TlsAcceptor;
@@ -14,10 +15,13 @@ use super::proxy::ProxyStream;
 use crate::cert::{Cert, CertFacade};
 use crate::util::to_u64;
 
-pub(super) fn wrap(
-    listener: impl Stream<Item = IoResult<ProxyStream>> + Send,
+pub(super) fn wrap<L, I>(
+    listener: L,
     pool: PgPool,
 ) -> impl Stream<Item = Result<impl AsyncRead + AsyncWrite + Send + Unpin + 'static, Error>> + Send
+where
+    L: Stream<Item = IoResult<I>> + Send,
+    I: Future<Output = IoResult<ProxyStream>> + Send,
 {
     let acceptor = Acceptor::new(pool);
 
@@ -26,13 +30,12 @@ pub(super) fn wrap(
         .zip(repeat(acceptor))
         .map(|(conn, acceptor)| conn.map(|c| (c, acceptor)))
         .map_ok(|(conn, acceptor)| async move {
-            let tls = acceptor.load_cert().await?;
+            let (conn, tls) = tokio::try_join!(conn.err_into(), acceptor.load_cert())?;
             Ok(tls.accept(conn).await?)
         })
         .try_buffer_unordered(100)
         .inspect_err(|err| error!("Stream error: {:?}", err))
         .filter(|stream| futures_util::future::ready(stream.is_ok()))
-        .into_stream()
 }
 
 struct Acceptor {