src/server/worker_task.rs: implement task control socket

[proxmox-backup.git] / src / bin / proxmox-backup-proxy.rs

diff --git a/src/bin/proxmox-backup-proxy.rs b/src/bin/proxmox-backup-proxy.rs
index 7eca6f173eea88a1faed2331a6e1de15b6fb9516..9661cbe1731be58b842bdc0c9f3e16072cd5a083 100644 (file)
--- a/src/bin/proxmox-backup-proxy.rs
+++ b/src/bin/proxmox-backup-proxy.rs
@@ -1,16 +1,17 @@
-#[macro_use]
-extern crate proxmox_backup;
-
+use proxmox_backup::try_block;
+use proxmox_backup::configdir;
 use proxmox_backup::tools;
-use proxmox_backup::api::router::*;
-use proxmox_backup::api::config::*;
+use proxmox_backup::server;
+use proxmox_backup::tools::daemon;
+use proxmox_backup::api_schema::router::*;
+use proxmox_backup::api_schema::config::*;
 use proxmox_backup::server::rest::*;
 use proxmox_backup::auth_helpers::*;
 
 use failure::*;
 use lazy_static::lazy_static;
 
-use futures::future::Future;
+use futures::*;
 use futures::stream::Stream;
 
 use hyper;
@@ -24,7 +25,6 @@ fn main() {
 }
 
 fn run() -> Result<(), Error> {
-
     if let Err(err) = syslog::init(
         syslog::Facility::LOG_DAEMON,
         log::LevelFilter::Info,
@@ -63,32 +63,57 @@ fn run() -> Result<(), Error> {
         Err(err) => bail!("unabled to decode pkcs12 identity {} - {}", cert_path, err),
     };
 
-    let addr = ([0,0,0,0,0,0,0,0], 8007).into();
-    let listener = tokio::net::TcpListener::bind(&addr)?;
-    let acceptor = native_tls::TlsAcceptor::new(identity)?;
-    let acceptor = std::sync::Arc::new(tokio_tls::TlsAcceptor::from(acceptor));
-    let connections = listener
-        .incoming()
-        .map_err(|e| Error::from(e))
-        .and_then(move |sock| acceptor.accept(sock).map_err(|e| e.into()))
-        .then(|r| match r {
-            // accept()s can fail here with an Err() when eg. the client rejects
-            // the cert and closes the connection, so we follow up with mapping
-            // it to an option and then filtering None with filter_map
-            Ok(c) => Ok::<_, Error>(Some(c)),
-            Err(_) => Ok(None),
-        })
-        .filter_map(|r| {
-            // Filter out the Nones
-            r
+    let server = daemon::create_daemon(
+        ([0,0,0,0,0,0,0,0], 8007).into(),
+        |listener| {
+            let acceptor = native_tls::TlsAcceptor::new(identity)?;
+            let acceptor = std::sync::Arc::new(tokio_tls::TlsAcceptor::from(acceptor));
+            let connections = listener
+                .incoming()
+                .map_err(Error::from)
+                .and_then(move |sock| acceptor.accept(sock).map_err(|e| e.into()))
+                .then(|r| match r {
+                    // accept()s can fail here with an Err() when eg. the client rejects
+                    // the cert and closes the connection, so we follow up with mapping
+                    // it to an option and then filtering None with filter_map
+                    Ok(c) => Ok::<_, Error>(Some(c)),
+                    Err(e) => {
+                        if let Some(_io) = e.downcast_ref::<std::io::Error>() {
+                            // "real" IO errors should not simply be ignored
+                            bail!("shutting down...");
+                        } else {
+                            // handshake errors just get filtered by filter_map() below:
+                            Ok(None)
+                        }
+                    }
+                })
+                .filter_map(|r| {
+                    // Filter out the Nones
+                    r
+                });
+            Ok(hyper::Server::builder(connections)
+                .serve(rest_server)
+                .map_err(|e| eprintln!("server error: {}", e))
+            )
+        },
+    )?;
+
+    tokio::run(lazy(||  {
+
+        let init_result: Result<(), Error> = try_block!({
+            server::create_task_control_socket()?;
+            server::server_state_init()?;
+            Ok(())
         });
 
-    let server = hyper::Server::builder(connections)
-        .serve(rest_server)
-        .map_err(|e| eprintln!("server error: {}", e));
+        if let Err(err) = init_result {
+            eprintln!("unable to start daemon - {}", err);
+        } else {
+            tokio::spawn(server);
+        }
 
-    // Run this server for... forever!
-    hyper::rt::run(server);
+        Ok(())
+    }));
 
     Ok(())
 }