feat: graceful shutdown (#23)

* feat: graceful shutdown

* feat: async net io

* chore: graceful shutdown

* chore: graceful shutdown

* chore: unix graceful shutdown

* chore: simplify code

* chore: child token

* chore: use child token to NOT kill calling server

* fix: missing tokio feature

* chore:  clippy

* chore: apply peer review feedback

abci/Cargo.toml

@@ -11,11 +11,18 @@ description = """tenderdash-abci provides a simple framework with which to build
 low-level applications on top of Tenderdash."""
 
 [features]
-default = ["server", "docker-tests", "crypto"]
+default = ["server", "docker-tests", "crypto", "tcp", "unix"]
 # docker-tests includes integration tests that require docker to be available
 docker-tests = ["server"]
-server = ["tracing-subscriber/fmt"]
+server = [
+    "tracing-subscriber/fmt",
+    "dep:tokio",
+    "dep:tokio-util",
+    "dep:futures",
+]
 crypto = ["dep:lhash"]
+tcp = ["server"]
+unix = ["server"]
 
 [[example]]
 name = "echo_socket"
@@ -26,12 +33,27 @@ tenderdash-proto = { version = "0.12.0-dev.2", path = "../proto" }
 bytes = { version = "1.0" }
 prost = { version = "0.11" }
 tracing = { version = "0.1", default-features = false }
-tracing-subscriber = { version = "0.3", optional = true, default-features = false }
+tracing-subscriber = { version = "0.3", optional = true, default-features = false, features = [
+    "ansi",
+    "env-filter",
+] }
 thiserror = "1.0.39"
 url = { version = "2.3.1" }
 semver = { version = "1.0.17" }
 lhash = { version = "1.0.1", features = ["sha256"], optional = true }
 hex = { version = "0.4" }
+tokio-util = { version = "0.7.8", features = [
+    "net",
+    "codec",
+], default-features = false, optional = true }
+tokio = { version = "1.28", features = [
+    "net",
+    "io-util",
+    "rt-multi-thread",
+    "sync",
+    "macros",
+], default-features = false, optional = true }
+futures = { version = "0.3.28", optional = true }
 
 [dev-dependencies]
 anyhow = "1.0.69"
@@ -41,3 +63,4 @@ bollard = { version = "0.14.0" }
 futures = { version = "0.3.26" }
 tokio = { version = "1", features = ["macros", "signal", "time", "io-std"] }
 hex = { version = "0.4" }
+lazy_static = { version = "1.4" }

abci/examples/echo_socket.rs

@@ -1,9 +1,12 @@
-use tenderdash_abci::{proto::abci, start_server, Application};
+use lazy_static::lazy_static;
+use tenderdash_abci::{proto::abci, Application, CancellationToken, ServerBuilder};
 use tracing::info;
 use tracing_subscriber::filter::LevelFilter;
 
 const SOCKET: &str = "/tmp/abci.sock";
-
+lazy_static! {
+    static ref CANCEL_TOKEN: CancellationToken = CancellationToken::new();
+}
 pub fn main() {
     let log_level = LevelFilter::DEBUG;
     tracing_subscriber::fmt().with_max_level(log_level).init();
@@ -12,9 +15,16 @@ pub fn main() {
     info!("This application listens on {SOCKET} and waits for incoming Tenderdash requests.");
 
     let socket = format!("unix://{}", SOCKET);
-    let server = start_server(&socket, EchoApp {}).expect("server failed");
+    let app = EchoApp {};
+
+    let cancel = CANCEL_TOKEN.clone();
+    let server = ServerBuilder::new(app, &socket)
+        .with_cancel_token(cancel)
+        .build()
+        .expect("server failed");
+
     loop {
-        match server.handle_connection() {
+        match server.next_client() {
             Ok(_) => {},
             Err(e) => tracing::error!("error {}", e),
         };
@@ -30,7 +40,11 @@ impl Application for EchoApp {
         &self,
         request: abci::RequestEcho,
     ) -> Result<abci::ResponseEcho, abci::ResponseException> {
-        info!("received echo");
+        info!("received echo, cancelling");
+
+        let cancel = CANCEL_TOKEN.clone();
+        cancel.cancel();
+
         Ok(abci::ResponseEcho {
             message: request.message,
         })

abci/src/lib.rs

@@ -20,7 +20,8 @@ use std::io;
 
 pub use application::{check_version, Application, RequestDispatcher};
 use prost::{DecodeError, EncodeError};
-pub use server::{start_server, Server};
+#[allow(deprecated)]
+pub use server::{start_server, CancellationToken, Server, ServerBuilder};
 pub use tenderdash_proto as proto;
 
 #[cfg(feature = "crypto")]
@@ -39,4 +40,8 @@ pub enum Error {
     Encode(#[from] EncodeError),
     #[error("cannot create canonical message: {0}")]
     Canonical(String),
+    #[error("server terminated")]
+    Cancelled(),
+    #[error("async runtime error")]
+    Async(String),
 }