Use a streaming BoxedBody

src/error.rs

@@ -1,3 +1,4 @@
+use std::convert::Infallible;
 use std::error::Error as StdError;
 use std::panic::Location;
 
@@ -79,6 +80,9 @@ pub enum ChorusError {
     // Hyper
     Hyper(hyper::Error),
 
+    // Infallible
+    Infallible,
+
     // Invalid URI
     InvalidUri(hyper::http::uri::InvalidUri),
 
@@ -182,6 +186,7 @@ impl std::fmt::Display for ChorusError {
             ChorusError::General(s) => write!(f, "{s}"),
             ChorusError::Http(e) => write!(f, "{e}"),
             ChorusError::Hyper(e) => write!(f, "{e}"),
+            ChorusError::Infallible => panic!("INFALLIBLE"),
             ChorusError::InvalidUri(e) => write!(f, "{e}"),
             ChorusError::InvalidUriParts(e) => write!(f, "{e}"),
             ChorusError::Io(e) => write!(f, "{e}"),
@@ -268,6 +273,7 @@ impl ChorusError {
             ChorusError::General(_) => 0.0,
             ChorusError::Http(_) => 0.0,
             ChorusError::Hyper(_) => 0.0,
+            ChorusError::Infallible => panic!("INFALLIBLE"),
             ChorusError::InvalidUri(_) => 0.0,
             ChorusError::InvalidUriParts(_) => 0.0,
             ChorusError::Io(_) => 0.0,
@@ -524,3 +530,9 @@ impl From<base64::DecodeError> for Error {
         }
     }
 }
+
+impl From<Infallible> for Error {
+    fn from(_: Infallible) -> Self {
+        panic!("INFALLIBLE")
+    }
+}

src/lib.rs

@@ -14,7 +14,8 @@ use crate::globals::GLOBALS;
 use crate::ip::{HashedIp, HashedPeer, IpData, SessionExit};
 use crate::reply::NostrReply;
 use futures::{sink::SinkExt, stream::StreamExt};
-use http_body_util::Full;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Empty};
 use hyper::body::{Bytes, Incoming};
 use hyper::service::Service;
 use hyper::upgrade::Upgraded;
@@ -78,7 +79,7 @@ struct ChorusService {
 }
 
 impl Service<Request<Incoming>> for ChorusService {
-    type Response = Response<Full<Bytes>>;
+    type Response = Response<BoxBody<Bytes, Self::Error>>;
     type Error = Error;
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
 
@@ -134,7 +135,7 @@ impl Service<Request<Incoming>> for ChorusService {
 async fn handle_http_request(
     peer: HashedPeer,
     mut request: Request<Incoming>,
-) -> Result<Response<Full<Bytes>>, Error> {
+) -> Result<Response<BoxBody<Bytes, Error>>, Error> {
     let ua = match request.headers().get("user-agent") {
         Some(ua) => ua.to_str().unwrap_or("NON-UTF8-HEADER").to_owned(),
         None => "(no user-agent)".to_owned(),
@@ -150,7 +151,7 @@ async fn handle_http_request(
         if *cur.value() >= max_conn {
             return Ok(Response::builder()
                 .status(StatusCode::TOO_MANY_REQUESTS)
-                .body(Full::new(Bytes::new()))?);
+                .body(Empty::new().map_err(|e| e.into()).boxed())?);
         }
     }
 
@@ -172,7 +173,7 @@ async fn handle_http_request(
             if !we_can_do_nostr {
                 return Ok(Response::builder()
                     .status(StatusCode::NOT_IMPLEMENTED)
-                    .body(Full::new(Bytes::new()))?);
+                    .body(Empty::new().map_err(|e| e.into()).boxed())?);
             }
         }
 
@@ -198,7 +199,7 @@ async fn handle_http_request(
         // Start the websocket thread
         tokio::spawn(async move { websocket_thread(peer, websocket, origin, ua).await });
 
-        Ok(response)
+        Ok(response.map(|body| body.map_err(|e| e.into()).boxed()))
     } else {
         web::serve_http(peer, request).await
     }

src/web/management/mod.rs

@@ -1,14 +1,18 @@
 use crate::error::{ChorusError, Error};
 use crate::globals::GLOBALS;
 use crate::ip::HashedPeer;
-use http_body_util::Full;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Full};
 use hyper::body::{Bytes, Incoming};
 use hyper::{Request, Response, StatusCode};
 use pocket_types::{Id, Pubkey};
 use serde_json::{json, Map, Value};
 mod auth;
 
-fn respond(json: serde_json::Value, status: StatusCode) -> Result<Response<Full<Bytes>>, Error> {
+fn respond(
+    json: serde_json::Value,
+    status: StatusCode,
+) -> Result<Response<BoxBody<Bytes, Error>>, Error> {
     let s: String = serde_json::to_string(&json)?;
     let response = Response::builder()
         .header("Access-Control-Allow-Origin", "*")
@@ -16,14 +20,18 @@ fn respond(json: serde_json::Value, status: StatusCode) -> Result<Response<Full<
         .header("Access-Control-Allow-Methods", "*")
         .header("Content-Type", "application/nostr+json")
         .status(status)
-        .body(s.into_bytes().into())?;
+        .body(
+            Full::new(s.into_bytes().into())
+                .map_err(|e| e.into())
+                .boxed(),
+        )?;
     Ok(response)
 }
 
 pub async fn handle(
     _peer: HashedPeer,
     request: Request<Incoming>,
-) -> Result<Response<Full<Bytes>>, Error> {
+) -> Result<Response<BoxBody<Bytes, Error>>, Error> {
     let command: Value = match auth::check_auth(request).await {
         Ok(v) => v,
         Err(e) => {

src/web/mod.rs

@@ -3,14 +3,15 @@ mod nip11;
 
 use crate::error::Error;
 use crate::ip::HashedPeer;
-use http_body_util::Full;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Full};
 use hyper::body::{Bytes, Incoming};
 use hyper::{Request, Response, StatusCode};
 
 pub async fn serve_http(
     peer: HashedPeer,
     request: Request<Incoming>,
-) -> Result<Response<Full<Bytes>>, Error> {
+) -> Result<Response<BoxBody<Bytes, Error>>, Error> {
     // check for Accept header of application/nostr+json
     if let Some(accept) = request.headers().get("Accept") {
         if let Ok(s) = accept.to_str() {
@@ -29,6 +30,10 @@ pub async fn serve_http(
         .header("Access-Control-Allow-Headers", "*")
         .header("Access-Control-Allow-Methods", "*")
         .status(StatusCode::OK)
-        .body("This is a nostr relay. Please use a nostr client to connect.".into())?;
+        .body(
+            Full::new("This is a nostr relay. Please use a nostr client to connect.".into())
+                .map_err(|e| e.into())
+                .boxed(),
+        )?;
     Ok(response)
 }

src/web/nip11.rs

@@ -2,11 +2,12 @@ use crate::config::Config;
 use crate::error::Error;
 use crate::globals::GLOBALS;
 use crate::ip::HashedPeer;
-use http_body_util::Full;
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Full};
 use hyper::body::Bytes;
 use hyper::{Response, StatusCode};
 
-pub async fn serve_nip11(peer: HashedPeer) -> Result<Response<Full<Bytes>>, Error> {
+pub async fn serve_nip11(peer: HashedPeer) -> Result<Response<BoxBody<Bytes, Error>>, Error> {
     log::debug!(target: "Client", "{}: sent NIP-11", peer);
     let rid = {
         let config = &*GLOBALS.config.read();
@@ -19,7 +20,7 @@ pub async fn serve_nip11(peer: HashedPeer) -> Result<Response<Full<Bytes>>, Erro
         .header("Access-Control-Allow-Methods", "*")
         .header("Content-Type", "application/nostr+json")
         .status(StatusCode::OK)
-        .body(rid.clone().into())?;
+        .body(Full::new(rid.clone().into()).map_err(|e| e.into()).boxed())?;
     Ok(response)
 }