fixed merge conflicts

idky137 · Jun 6, 2024 · 9c73ccd · 9c73ccd
2 parents b1b09d3 + b119bf9
commit 9c73ccd
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 14 deletions.
diff --git a/zingo-proxyd/src/bin/zingoproxyd.rs b/zingo-proxyd/src/bin/zingoproxyd.rs
@@ -7,11 +7,8 @@ use std::{
         Arc,
     },
 };
-
 use zingoproxylib::proxy::spawn_proxy;
 
-use ctrlc;
-
 #[tokio::main]
 async fn main() {
     let online = Arc::new(AtomicBool::new(true));
@@ -43,5 +40,8 @@ async fn main() {
     )
     .await;
 
-    while online.load(Ordering::SeqCst) {}
+    let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500));
+    while online.load(Ordering::SeqCst) {
+        interval.tick().await;
+    }
 }
diff --git a/zingo-proxyd/src/nym_server.rs b/zingo-proxyd/src/nym_server.rs
@@ -28,9 +28,16 @@ impl NymServer {
     ) -> tokio::task::JoinHandle<Result<(), tonic::transport::Error>> {
         let mut request_in: Vec<ReconstructedMessage> = Vec::new();
         tokio::task::spawn(async move {
+            // NOTE: This interval may need to be reduced or removed / moved once scale testing begins.
+            let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(50));
             while online.load(Ordering::SeqCst) {
                 while let Some(request_nym) = self.0 .0.wait_for_messages().await {
                     if request_nym.is_empty() {
+                        interval.tick().await;
+                        if !online.load(Ordering::SeqCst) {
+                            println!("Nym server shutting down.");
+                            return Ok(());
+                        }
                         continue;
                     }
                     request_in = request_nym;
@@ -74,6 +81,7 @@ impl NymServer {
                     .await
                     .unwrap();
             }
+            println!("Nym server shutting down.");
             Ok(())
         })
     }

diff --git a/zingo-proxyd/src/server.rs b/zingo-proxyd/src/server.rs
@@ -28,18 +28,37 @@ impl ProxyServer {
         tokio::task::spawn(async move {
             let svc = CompactTxStreamerServer::new(self.0);
             let sockaddr = SocketAddr::new(std::net::IpAddr::V4(Ipv4Addr::LOCALHOST), port.into());
-            println!("@zingoproxyd: GRPC server listening on: {sockaddr}.");
-            while online.load(Ordering::SeqCst) {
-                let server = tonic::transport::Server::builder()
-                    .add_service(svc.clone())
-                    .serve(sockaddr)
-                    .await;
-                match server {
-                    Ok(_) => (),
-                    Err(e) => return Err(e),
+            println!("@zingoproxyd: gRPC server listening on: {sockaddr}");
+
+            let server = tonic::transport::Server::builder()
+                .add_service(svc.clone())
+                .serve(sockaddr);
+
+            let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(500));
+            tokio::select! {
+                result = server => {
+                    match result {
+                        Ok(_) => {
+                            // TODO: Gracefully restart gRPC server.
+                            println!("@zingoproxyd: gRPC Server closed early. Restart required");
+                            Ok(())
+                            }
+                        Err(e) => {
+                            // TODO: restart server or set online to false and exit
+                            println!("@zingoproxyd: gRPC Server closed with error: {}. Restart required", e);
+                            Err(e)
+                            }
+                    }
+                }
+                _ = async {
+                    while online.load(Ordering::SeqCst) {
+                        interval.tick().await;
+                    }
+                } => {
+                    println!("@zingoproxyd: gRPC server shutting down.");
+                    Ok(())
                 }
             }
-            Ok(())
         })
     }