From 35687652e7956a41c98d8ab88c5f545dc088975a Mon Sep 17 00:00:00 2001
From: Brandon Thomas <echelon@gmail.com>
Date: Mon, 7 Sep 2020 16:04:26 -0400
Subject: [PATCH] proxying works in a rough first pass

---
 src/main.rs        | 43 ++++++++++++++++++++++++++++++++++++-------
 src/speak_proxy.rs | 34 +++++++++++++++++++++++++---------
 2 files changed, 61 insertions(+), 16 deletions(-)

diff --git a/src/main.rs b/src/main.rs
index db5a406..764cf59 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -8,6 +8,7 @@ pub mod speak_proxy;
 use anyhow::Result as AnyhowResult;
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
 use futures::future::{self, Future};
+use crate::speak_proxy::speak_proxy;
 use hyper::{Method, StatusCode};
 use hyper::header::HeaderValue;
 use hyper::http::header::HeaderName;
@@ -133,10 +134,34 @@ pub async fn handle(
 {
   match req.uri().path() {
     "/speak" => {
-      debug_request(req)
+      let nr_transaction = newrelic_logger.web_transaction("/speak");
+      let result = speak_proxy(client_ip, req, router, nr_transaction, "/speak").await;
+      match result {
+        Ok(response) => {
+          Ok(response)
+        },
+        Err(_error) => {
+          Ok(Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Body::empty())
+            .unwrap())
+        }
+      }
     },
     "/speak_spectrogram" => {
-      debug_request(req)
+      let nr_transaction = newrelic_logger.web_transaction("/speak_spectrogram");
+      let result = speak_proxy(client_ip, req, router, nr_transaction, "/speak_spectrogram").await;
+      match result {
+        Ok(response) => {
+          Ok(response)
+        },
+        Err(_error) => {
+          Ok(Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Body::empty())
+            .unwrap())
+        }
+      }
     },
     "/proxy_health" => {
       debug_request(req)
@@ -149,11 +174,15 @@ pub async fn handle(
       info!("Forwarding to `{}` random host: {}", &forward, req.uri());
 
       match hyper_reverse_proxy::call(client_ip, &forward, req).await {
-        Ok(response) => {Ok(response)}
-        Err(error) => {Ok(Response::builder()
-          .status(StatusCode::INTERNAL_SERVER_ERROR)
-          .body(Body::empty())
-          .unwrap())}
+        Ok(response) => {
+          Ok(response)
+        },
+        Err(error) => {
+          Ok(Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Body::empty())
+            .unwrap())
+        }
       }
     },
   }
diff --git a/src/speak_proxy.rs b/src/speak_proxy.rs
index 52b5132..076600b 100644
--- a/src/speak_proxy.rs
+++ b/src/speak_proxy.rs
@@ -18,6 +18,7 @@ use std::str::FromStr;
 use std::sync::Arc;
 use std::convert::Infallible;
 use hyper_reverse_proxy::ProxyError;
+use futures::TryStreamExt;
 
 /// NB: This much match the shape of SpeakRequest in the 'voder/tts_service' code.
 /// This is used for both /speak and /speak_spectrogram requests.
@@ -59,17 +60,22 @@ struct RequestDetails {
     .boxed()
 }*/
 
+#[derive(Serialize, Debug)]
+pub struct ResponseError {
+  pub reason: String,
+}
+
 pub async fn speak_proxy(
-  remote_addr: SocketAddr,
-  req: Request<Vec<u8>>,
+  ip_addr: IpAddr,
+  req: Request<Body>,
   router: Arc<Router>,
   transaction: MaybeNewRelicTransaction,
-  endpoint: &'static str) -> Box<Future<Output=Result<Response<Body>, ProxyError>>>
+  endpoint: &'static str) -> Result<Response<Body>, ResponseError>
 {
   let mut headers = req.headers().clone();
 
   // This is the IP of the Digital Ocean load balancer.
-  let gateway_ip = HeaderValue::from_str(&remote_addr.ip().to_string())
+  let gateway_ip = HeaderValue::from_str(&ip_addr.to_string())
     .ok()
     .unwrap_or(HeaderValue::from_static(""));
 
@@ -89,9 +95,17 @@ pub async fn speak_proxy(
   headers.insert(HeaderName::from_static("x-voder-proxy-for"), upstream_client_ip.clone());
   headers.insert(HeaderName::from_static("x-gateway-ip"), gateway_ip);
 
-  let request_bytes = req.body();
+  let (parts, body) = req.into_parts();
 
-  let request = serde_json::from_slice::<SpeakRequest>(request_bytes)
+  let request_bytes = body
+    .try_fold(Vec::new(), |mut data, chunk| async move {
+      data.extend_from_slice(&chunk);
+      Ok(data)
+    })
+    .await
+    .unwrap();
+
+  let request = serde_json::from_slice::<SpeakRequest>(&request_bytes)
     .unwrap();
 
   let request_details = RequestDetails {
@@ -119,9 +133,11 @@ pub async fn speak_proxy(
     .unwrap();
 
   //Box::new(future::ok(Vec::new()))
-  match hyper_reverse_proxy::call(remote_addr.ip(), &proxy_host, new_req).await {
-    Ok(result) => Box::new(future::ok(result)),
-    Err(result) => Box::new(future::err(result)),
+  match hyper_reverse_proxy::call(ip_addr, &proxy_host, new_req).await {
+    Ok(result) => Ok(result),
+    Err(_result) => Err(ResponseError {
+      reason: "Uknown failure".to_string()
+    }),
   }
 }