Feat: listen for unix signal

libs/hbb_common/build.rs

@@ -5,10 +5,7 @@ fn main() {
         .out_dir("src/protos")
         .inputs(&["protos/rendezvous.proto", "protos/message.proto"])
         .include("protos")
-        .customize(
+        .customize(protobuf_codegen::Customize::default().tokio_bytes(true))
-            protobuf_codegen::Customize::default()
-            .tokio_bytes(true)
-        )
         .run()
         .expect("Codegen failed.");
 }

libs/hbb_common/src/fs.rs

@@ -565,4 +565,3 @@ pub fn transform_windows_path(entries: &mut Vec<FileEntry>) {
         entry.name = entry.name.replace("\\", "/");
     }
 }
-

src/common.rs

@@ -1,5 +1,8 @@
 use clap::App;
-use hbb_common::{anyhow::Context, log, ResultType};
+use hbb_common::{
+    anyhow::{Context, Result},
+    log, ResultType,
+};
 use ini::Ini;
 use sodiumoxide::crypto::sign;
 use std::{
@@ -138,3 +141,43 @@ pub fn gen_sk(wait: u64) -> (String, Option<sign::SecretKey>) {
     }
     ("".to_owned(), None)
 }
+
+#[cfg(unix)]
+pub async fn listen_signal() -> Result<()> {
+    use hbb_common::tokio;
+    use hbb_common::tokio::signal::unix::{signal, SignalKind};
+
+    tokio::spawn(async {
+        let mut s = signal(SignalKind::hangup())?;
+        let hangup = s.recv();
+        let mut s = signal(SignalKind::terminate())?;
+        let terminate = s.recv();
+        let mut s = signal(SignalKind::interrupt())?;
+        let interrupt = s.recv();
+        let mut s = signal(SignalKind::quit())?;
+        let quit = s.recv();
+
+        tokio::select! {
+            _ = hangup => {
+                log::info!("signal hangup");
+            }
+            _ = terminate => {
+                log::info!("signal terminate");
+            }
+            _ = interrupt => {
+                log::info!("signal interrupt");
+            }
+            _ = quit => {
+                log::info!("signal quit");
+            }
+        }
+        Ok(())
+    })
+    .await?
+}
+
+#[cfg(not(unix))]
+pub async fn listen_signal() -> Result<()> {
+    let () = std::future::pending().await;
+    unreachable!();
+}

src/peer.rs

@@ -1,10 +1,10 @@
 use crate::common::*;
 use crate::database;
 use hbb_common::{
+    bytes::Bytes,
     log,
     rendezvous_proto::*,
     tokio::sync::{Mutex, RwLock},
-    bytes::Bytes,
     ResultType,
 };
 use serde_derive::{Deserialize, Serialize};

src/relay_server.rs

@@ -81,15 +81,22 @@ pub async fn start(port: &str, key: &str) -> ResultType<()> {
     log::info!("Listening on tcp {}", addr);
     let addr2 = format!("0.0.0.0:{}", port.parse::<u16>().unwrap() + 2);
     log::info!("Listening on websocket {}", addr2);
-    loop {
+    let main_task = async move {
-        log::info!("Start");
+        loop {
-        io_loop(
+            log::info!("Start");
-            new_listener(&addr, false).await?,
+            io_loop(
-            new_listener(&addr2, false).await?,
+                new_listener(&addr, false).await?,
-            &key,
+                new_listener(&addr2, false).await?,
-        )
+                &key,
-        .await;
+            )
-    }
+            .await;
+        }
+    };
+    let listen_signal = crate::common::listen_signal();
+    tokio::select!(
+        res = main_task => res,
+        res = listen_signal => res,
+    )
 }
 
 fn check_params() {

src/rendezvous_server.rs

@@ -170,37 +170,44 @@ impl RendezvousServer {
                 allow_err!(test_hbbs(test_addr).await);
             });
         };
-        loop {
+        let main_task = async move {
-            log::info!("Start");
+            loop {
-            match rs
+                log::info!("Start");
-                .io_loop(
+                match rs
-                    &mut rx,
+                    .io_loop(
-                    &mut listener,
+                        &mut rx,
-                    &mut listener2,
+                        &mut listener,
-                    &mut listener3,
+                        &mut listener2,
-                    &mut socket,
+                        &mut listener3,
-                    &key,
+                        &mut socket,
-                )
+                        &key,
-                .await
+                    )
-            {
+                    .await
-                LoopFailure::UdpSocket => {
+                {
-                    drop(socket);
+                    LoopFailure::UdpSocket => {
-                    socket = FramedSocket::new_with_buf_size(&addr, rmem).await?;
+                        drop(socket);
-                }
+                        socket = FramedSocket::new_with_buf_size(&addr, rmem).await?;
-                LoopFailure::Listener => {
+                    }
-                    drop(listener);
+                    LoopFailure::Listener => {
-                    listener = new_listener(&addr, false).await?;
+                        drop(listener);
-                }
+                        listener = new_listener(&addr, false).await?;
-                LoopFailure::Listener2 => {
+                    }
-                    drop(listener2);
+                    LoopFailure::Listener2 => {
-                    listener2 = new_listener(&addr2, false).await?;
+                        drop(listener2);
-                }
+                        listener2 = new_listener(&addr2, false).await?;
-                LoopFailure::Listener3 => {
+                    }
-                    drop(listener3);
+                    LoopFailure::Listener3 => {
-                    listener3 = new_listener(&addr3, false).await?;
+                        drop(listener3);
+                        listener3 = new_listener(&addr3, false).await?;
+                    }
                 }
             }
-        }
+        };
+        let listen_signal = listen_signal();
+        tokio::select!(
+            res = main_task => res,
+            res = listen_signal => res,
+        )
     }
 
     async fn io_loop(

src/version.rs

@@ -1 +1 @@
-pub const VERSION: &str = "1.1.6";
+pub const VERSION: &str = "1.1.6";