WIP upgraded hyper-reverse-proxy, hyper, and futures

src/main.rs

@@ -7,17 +7,17 @@ pub mod speak_proxy;
 
 use anyhow::Result as AnyhowResult;
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
-use crate::speak_proxy::{speak_proxy, speak_proxy_with_retry};
+use crate::speak_proxy::handle;
 use futures::future::{self, Future};
 use hyper::Method;
 use hyper::header::HeaderValue;
 use hyper::http::header::HeaderName;
-use hyper::rt::Stream;
 use hyper::server::conn::AddrStream;
 use hyper::service::{service_fn, make_service_fn};
 use hyper::{Body, Request, Response, Server};
 use rand::seq::IteratorRandom;
 use std::collections::HashMap;
+use std::convert::Infallible;
 use std::env;
 use std::fmt::Debug;
 use std::fmt::Display;
@@ -25,7 +25,8 @@ use std::net::SocketAddr;
 use std::str::FromStr;
 use std::sync::Arc;
 
-type BoxFut = Box<dyn Future<Item=Response<Body>, Error=hyper::Error> + Send>;
+//type BoxFut = Box<dyn Future + Send>;
+type BoxFut = Box<dyn Future<Output=Response<Body>> + Send>;
 
 const ENV_DEFAULT_ROUTE : &'static str = "DEFAULT_ROUTE";
 const ENV_NEWRELIC_API_KEY : &'static str = "NEWRELIC_API_KEY";
@@ -88,16 +89,16 @@ fn get_env_num<T>(env_name: &str, default: T) -> AnyhowResult<T>
   }
 }
 
-fn _debug_request(req: Request<Body>) -> BoxFut {
+/*fn _debug_request(req: Request<Body>) -> BoxFut {
   let body_str = format!("{:?}", req);
   let response = Response::new(Body::from(body_str));
   Box::new(future::ok(response))
-}
+}*/
 
-fn health_check_response(_req: Request<Body>) -> BoxFut {
+/*fn health_check_response(_req: Request<Body>) -> BoxFut {
   let response = Response::new(Body::from("healthy"));
   Box::new(future::ok(response))
-}
+}*/
 
 pub struct Router {
   pub routes: HashMap<String, String>,
@@ -120,7 +121,8 @@ impl Router {
   }
 }
 
-fn main() -> AnyhowResult<()> {
+#[tokio::main]
+async fn main() {
   if env::var(ENV_RUST_LOG)
     .as_ref()
     .ok()
@@ -134,7 +136,8 @@ fn main() -> AnyhowResult<()> {
   env_logger::init();
 
   let default_route = get_env_string(ENV_DEFAULT_ROUTE, DEFAULT_DEFAULT_ROUTE);
-  let service_port = get_env_num::<u16>(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT)?;
+  let service_port = get_env_num::<u16>(ENV_SERVICE_PORT, DEFAULT_SERVICE_PORT)
+    .expect("should have env var");
   let newrelic_api_key = get_env_string(ENV_NEWRELIC_API_KEY, DEFAULT_NEWRELIC_API_KEY);
 
   info!("Default route: {}", default_route);
@@ -142,7 +145,8 @@ fn main() -> AnyhowResult<()> {
   let proxy_configs_file = get_env_string(ENV_PROXY_CONFIG_FILE, DEFAULT_PROXY_CONFIG_FILE);
   info!("Proxy config file: {}", proxy_configs_file);
 
-  let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file)?;
+  let proxy_configs = ProxyConfigs::load_from_file(&proxy_configs_file)
+    .expect("should have configs");
   info!("Proxy configs: {:?}", proxy_configs);
 
   let mut route_map = HashMap::new();
@@ -165,7 +169,7 @@ fn main() -> AnyhowResult<()> {
 
   let newrelic_logger = Arc::new(newrelic_logger);
 
-  let make_service = make_service_fn(move |socket: &AddrStream| {
+  /*let make_service = make_service_fn(move |socket: &AddrStream| {
     let remote_addr = socket.remote_addr();
     info!("Got a request from: {:?}", remote_addr);
 
@@ -177,18 +181,18 @@ fn main() -> AnyhowResult<()> {
       let newrelic_logger3 = newrelic_logger2.clone();
 
       match (req.method(), req.uri().path()) {
-        (&Method::POST, "/speak") => {
+        /*(&Method::POST, "/speak") => {
           let nr_transaction = newrelic_logger3.web_transaction("/speak");
-          speak_proxy_with_retry(req, remote_addr.clone(), router3, nr_transaction, "/speak")
+          speak_proxy(req, remote_addr.clone(), router3, nr_transaction, "/speak")
         },
         (&Method::POST, "/speak_spectrogram") => {
           let nr_transaction = newrelic_logger3.web_transaction("/speak_spectrogram");
           speak_proxy(req, remote_addr.clone(), router3, nr_transaction, "/speak_spectrogram")
-        },
-        (&Method::GET, "/proxy_health") => {
+        },*/
+        /*(&Method::GET, "/proxy_health") => {
           let _droppable_transaction = newrelic_logger3.web_transaction("/proxy_health");
           health_check_response(req)
-        },
+        },*/
         _ => {
           let _droppable_transaction = newrelic_logger3.web_transaction("unmatched endpoint");
           let forward = router3.get_random_host();
@@ -197,17 +201,24 @@ fn main() -> AnyhowResult<()> {
         }
       }
     })
+  });*/
+
+  let make_service = make_service_fn(|conn: &AddrStream| {
+    let remote_addr = conn.remote_addr().ip();
+    async move {
+      Ok::<_, Infallible>(service_fn(move |req| handle(remote_addr, req)))
+    }
   });
 
   let addr = ([0, 0, 0, 0], service_port).into();
 
   info!("Running server on {:?}", addr);
 
-  let server = Server::bind(&addr)
-    .serve(make_service)
-    .map_err(|e| eprintln!("server error: {}", e));
+  let server = Server::bind(&addr).serve(make_service);
 
-  hyper::rt::run(server);
+  if let Err(e) = server.await {
+    eprintln!("server error: {}", e);
+  }
 
-  Ok(())
+  ()
 }

src/speak_proxy.rs

@@ -2,10 +2,9 @@ use anyhow::Result as AnyhowResult;
 use crate::{BoxFut, Router};
 use crate::newrelic_logger::{NewRelicLogger, MaybeNewRelicTransaction};
 use futures::future::{self, Future};
-use hyper::Method;
+use hyper::{Method, StatusCode};
 use hyper::header::HeaderValue;
 use hyper::http::header::HeaderName;
-use hyper::rt::Stream;
 use hyper::server::conn::AddrStream;
 use hyper::service::{service_fn, make_service_fn};
 use hyper::{Body, Request, Response, Server};
@@ -14,10 +13,10 @@ use std::collections::HashMap;
 use std::env;
 use std::fmt::Debug;
 use std::fmt::Display;
-use std::net::SocketAddr;
+use std::net::{SocketAddr, IpAddr};
 use std::str::FromStr;
 use std::sync::Arc;
-use futures::IntoFuture;
+use std::convert::Infallible;
 
 /// NB: This much match the shape of SpeakRequest in the 'voder/tts_service' code.
 /// This is used for both /speak and /speak_spectrogram requests.
@@ -34,7 +33,15 @@ struct RequestDetails {
   pub speaker: String,
 }
 
-pub fn speak_proxy_with_retry(
+/*pub fn copy_request(request: &Request<Body>) -> Request<Body> {
+  let body = hyper::body::to_bytes(request.body());
+  Request::builder()
+    .uri(request.uri())
+    .method(request.method())
+    .unwrap()
+}*/
+
+/*pub fn speak_proxy_with_retry(
   req: Request<Body>,
   remote_addr: SocketAddr,
   router: Arc<Router>,
@@ -49,9 +56,9 @@ pub fn speak_proxy_with_retry(
       }
     })
     .boxed()
-}
+}*/
 
-pub fn speak_proxy(
+/*pub fn speak_proxy(
   req: Request<Body>,
   remote_addr: SocketAddr,
   router: Arc<Router>,
@@ -112,4 +119,34 @@ pub fn speak_proxy(
 
     hyper_reverse_proxy::call(remote_addr.ip(), &proxy_host, new_req)
   }))
+}*/
+
+
+fn debug_request(req: Request<Body>) -> Result<Response<Body>, Infallible>  {
+  let body_str = format!("{:?}", req);
+  Ok(Response::new(Body::from(body_str)))
+}
+
+pub async fn handle(client_ip: IpAddr, req: Request<Body>) -> Result<Response<Body>, Infallible> {
+  if req.uri().path().starts_with("/target/first") {
+  // will forward requests to port 13901
+  match hyper_reverse_proxy::call(client_ip, "http://127.0.0.1:13901", req).await {
+    Ok(response) => {Ok(response)}
+    Err(error) => {Ok(Response::builder()
+      .status(StatusCode::INTERNAL_SERVER_ERROR)
+      .body(Body::empty())
+      .unwrap())}
+    }
+  } else if req.uri().path().starts_with("/target/second") {
+    // will forward requests to port 13902
+    match hyper_reverse_proxy::call(client_ip, "http://127.0.0.1:13902", req).await {
+      Ok(response) => {Ok(response)}
+      Err(error) => {Ok(Response::builder()
+        .status(StatusCode::INTERNAL_SERVER_ERROR)
+        .body(Body::empty())
+        .unwrap())}
+    }
+  } else {
+    debug_request(req)
+  }
 }