rpc: merge rpcserver2 to rpcserver & fix bins affected by the merging

Cargo.lock

@@ -1177,6 +1177,7 @@ dependencies = [
  "serde_json",
  "simplelog",
  "smol",
+ "url",
 ]
 
 [[package]]
@@ -4003,6 +4004,7 @@ dependencies = [
  "structopt",
  "structopt-toml",
  "thiserror",
+ "url",
 ]
 
 [[package]]

bin/dao-cli/src/main.rs

@@ -35,11 +35,11 @@ impl Client {
     }
 
     async fn request(&self, r: jsonrpc::JsonRequest) -> Result<Value> {
-        let reply: JsonResult =
-            match jsonrpc::send_request(&Url::parse(&self.url)?, json!(r)).await {
-                Ok(v) => v,
-                Err(e) => return Err(e),
-            };
+        let reply: JsonResult = match jsonrpc::send_request(&Url::parse(&self.url)?, json!(r)).await
+        {
+            Ok(v) => v,
+            Err(e) => return Err(e),
+        };
 
         match reply {
             JsonResult::Resp(r) => {

bin/daod/Cargo.toml

@@ -21,6 +21,7 @@ easy-parallel = "3.2.0"
 log = "0.4.16"
 num_cpus = "1.13.1"
 simplelog = "0.12.0"
+url = "2.2.2"
 
 # Encoding and parsing
 serde_json = "1.0.79"

bin/daod/src/main.rs

@@ -1,36 +1,24 @@
-use async_executor::Executor;
+use std::sync::Arc;
+
 use async_trait::async_trait;
+use log::debug;
+use serde_json::{json, Value};
+use simplelog::{ColorChoice, LevelFilter, TermLogger, TerminalMode};
+use url::Url;
+
 use darkfi::{
     rpc::{
         jsonrpc::{error as jsonerr, response as jsonresp, ErrorCode::*, JsonRequest, JsonResult},
-        rpcserver::{listen_and_serve, RequestHandler, RpcServerConfig},
+        rpcserver::{listen_and_serve, RequestHandler},
     },
-    util::expand_path,
     Result,
 };
-use easy_parallel::Parallel;
-use log::debug;
-use serde_json::{json, Value};
-use simplelog::{ColorChoice, LevelFilter, TermLogger, TerminalMode};
-use std::{
-    net::{IpAddr, Ipv4Addr, SocketAddr},
-    sync::Arc,
-};
-
-async fn start(executor: Arc<Executor<'_>>) -> Result<()> {
-    let rpc_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 7777);
-    let server_config = RpcServerConfig {
-        socket_addr: rpc_addr,
-        use_tls: false,
-        // this is all random filler that is meaningless bc tls is disabled
-        // TODO: cleanup
-        identity_path: expand_path("../..")?,
-        identity_pass: "test".to_string(),
-    };
 
+async fn start() -> Result<()> {
+    let rpc_addr = Url::parse("tcp://127.0.0.1:7777")?;
     let rpc_interface = Arc::new(JsonRpcInterface {});
 
-    listen_and_serve(server_config, rpc_interface, executor).await?;
+    listen_and_serve(rpc_addr, rpc_interface).await?;
     Ok(())
 }
 
@@ -38,7 +26,7 @@ struct JsonRpcInterface {}
 
 #[async_trait]
 impl RequestHandler for JsonRpcInterface {
-    async fn handle_request(&self, req: JsonRequest, _executor: Arc<Executor<'_>>) -> JsonResult {
+    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if req.params.as_array().is_none() {
             return JsonResult::Err(jsonerr(InvalidParams, None, req.id))
         }
@@ -62,10 +50,6 @@ impl JsonRpcInterface {
 
 #[async_std::main]
 async fn main() -> Result<()> {
-    //let args = CliDao::parse();
-
-    //let matches = CliDao::command().get_matches();
-
     TermLogger::init(
         LevelFilter::Debug,
         simplelog::Config::default(),
@@ -73,24 +57,6 @@ async fn main() -> Result<()> {
         ColorChoice::Auto,
     )?;
 
-    //let rpc_addr = "tcp:://127.0.0.1:7777";
-    //let client = Arc::new(Client::new(rpc_addr.to_string()));
-
-    let nthreads = num_cpus::get();
-    let (signal, shutdown) = async_channel::unbounded::<()>();
-    let ex = Arc::new(Executor::new());
-    //let ex2 = ex.clone();
-    let ex3 = ex.clone();
-    let (_, result) = Parallel::new()
-        .each(0..nthreads, |_| smol::future::block_on(ex.run(shutdown.recv())))
-        .finish(|| {
-            smol::future::block_on(async move {
-                start(ex3.clone()).await?;
-                //client.run_client(client.clone(), ex2.clone()).await?;
-                drop(signal);
-                Ok::<(), darkfi::Error>(())
-            })
-        });
-
-    result
+    start().await?;
+    Ok(())
 }

bin/darkfid2/src/main.rs

@@ -34,7 +34,7 @@ use darkfi::{
             ErrorCode::{InvalidParams, MethodNotFound},
             JsonRequest, JsonResult,
         },
-        rpcserver2::{listen_and_serve, RequestHandler},
+        rpcserver::{listen_and_serve, RequestHandler},
     },
     util::{
         cli::{log_config, spawn_config},

bin/faucetd/src/main.rs

@@ -33,7 +33,7 @@ use darkfi::{
             ErrorCode::{InternalError, InvalidParams, MethodNotFound},
             JsonRequest, JsonResult,
         },
-        rpcserver2::{listen_and_serve, RequestHandler},
+        rpcserver::{listen_and_serve, RequestHandler},
     },
     util::{
         cli::{log_config, spawn_config},

bin/ircd/src/main.rs

@@ -13,11 +13,12 @@ use log::{debug, error, info, warn};
 use simplelog::{ColorChoice, TermLogger, TerminalMode};
 use smol::future;
 use structopt_toml::StructOptToml;
+use url::Url;
 
 use darkfi::{
     async_daemonize, net,
     raft::{NetMsg, ProtocolRaft, Raft},
-    rpc::rpcserver::{listen_and_serve, RpcServerConfig},
+    rpc::rpcserver::listen_and_serve,
     util::{
         cli::{log_config, spawn_config},
         path::{expand_path, get_config_path},
@@ -167,17 +168,11 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     //
     // RPC interface
     //
-    let rpc_config = RpcServerConfig {
-        socket_addr: settings.rpc_listen,
-        use_tls: false,
-        identity_path: Default::default(),
-        identity_pass: Default::default(),
-    };
-    let executor_cloned = executor.clone();
-    let rpc_interface = Arc::new(JsonRpcInterface { addr: settings.rpc_listen, p2p: p2p.clone() });
-    let rpc_task = executor.spawn(async move {
-        listen_and_serve(rpc_config, rpc_interface, executor_cloned.clone()).await
-    });
+    let rpc_listen_addr = Url::parse(&settings.rpc_listen)?;
+    let rpc_interface =
+        Arc::new(JsonRpcInterface { addr: rpc_listen_addr.clone(), p2p: p2p.clone() });
+    let rpc_task =
+        executor.spawn(async move { listen_and_serve(rpc_listen_addr, rpc_interface).await });
 
     //
     // IRC instance

bin/ircd/src/rpc.rs

@@ -1,9 +1,9 @@
-use std::{net::SocketAddr, sync::Arc};
+use std::sync::Arc;
 
-use async_executor::Executor;
 use async_trait::async_trait;
 use log::debug;
 use serde_json::{json, Value};
+use url::Url;
 
 use darkfi::{
     net,
@@ -15,13 +15,13 @@ use darkfi::{
 };
 
 pub struct JsonRpcInterface {
-    pub addr: SocketAddr,
+    pub addr: Url,
     pub p2p: net::P2pPtr,
 }
 
 #[async_trait]
 impl RequestHandler for JsonRpcInterface {
-    async fn handle_request(&self, req: JsonRequest, _executor: Arc<Executor<'_>>) -> JsonResult {
+    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if req.params.as_array().is_none() {
             return jsonrpc::error(ErrorCode::InvalidRequest, None, req.id).into()
         }

bin/ircd/src/settings.rs

@@ -18,8 +18,8 @@ pub struct Args {
     #[structopt(long)]
     pub config: Option<String>,
     /// JSON-RPC listen URL
-    #[structopt(long = "rpc", default_value = "127.0.0.1:11055")]
-    pub rpc_listen: SocketAddr,
+    #[structopt(long = "rpc", default_value = "tcp://127.0.0.1:11055")]
+    pub rpc_listen: String,
     /// IRC listen URL
     #[structopt(long = "irc", default_value = "127.0.0.1:11066")]
     pub irc_listen: SocketAddr,

bin/tau/taud/Cargo.toml

@@ -26,6 +26,7 @@ rand = "0.8.5"
 chrono = "0.4.19"
 thiserror = "1.0.30"
 ctrlc-async = {version= "3.2.2", default-features = false, features = ["async-std", "termination"]}
+url = "2.2.2"
 
 # Encoding and parsing
 serde = {version = "1.0.136", features = ["derive"]}

bin/tau/taud/src/jsonrpc.rs

@@ -39,7 +39,7 @@ struct BaseTaskInfo {
 
 #[async_trait]
 impl RequestHandler for JsonRpcInterface {
-    async fn handle_request(&self, req: JsonRequest, _executor: Arc<Executor<'_>>) -> JsonResult {
+    async fn handle_request(&self, req: JsonRequest) -> JsonResult {
         if req.params.as_array().is_none() {
             return JsonResult::Err(jsonerr(ErrorCode::InvalidParams, None, req.id))
         }

bin/tau/taud/src/main.rs

@@ -9,11 +9,12 @@ use log::{debug, error, info, warn};
 use simplelog::{ColorChoice, TermLogger, TerminalMode};
 use smol::future;
 use structopt_toml::StructOptToml;
+use url::Url;
 
 use darkfi::{
     async_daemonize, net,
     raft::{NetMsg, ProtocolRaft, Raft},
-    rpc::rpcserver::{listen_and_serve, RpcServerConfig},
+    rpc::rpcserver::listen_and_serve,
     util::{
         cli::{log_config, spawn_config},
         expand_path,
@@ -124,13 +125,6 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
     //
     // RPC
     //
-    let server_config = RpcServerConfig {
-        socket_addr: settings.rpc_listen,
-        use_tls: false,
-        // this is all random filler that is meaningless bc tls is disabled
-        identity_path: Default::default(),
-        identity_pass: Default::default(),
-    };
 
     let (rpc_snd, rpc_rcv) = async_channel::unbounded::<Option<TaskInfo>>();
 
@@ -138,8 +132,9 @@ async fn realmain(settings: Args, executor: Arc<Executor<'_>>) -> Result<()> {
         Arc::new(JsonRpcInterface::new(rpc_snd, datastore_path.clone(), nickname.unwrap()));
 
     let executor_cloned = executor.clone();
+    let rpc_listener_url = Url::parse(&settings.rpc_listen)?;
     let rpc_listener_task =
-        executor_cloned.spawn(listen_and_serve(server_config, rpc_interface, executor.clone()));
+        executor_cloned.spawn(listen_and_serve(rpc_listener_url, rpc_interface));
 
     let net_settings = settings.net;
 

bin/tau/taud/src/settings.rs

@@ -18,8 +18,8 @@ pub struct Args {
     #[structopt(long)]
     pub config: Option<String>,
     /// JSON-RPC listen URL
-    #[structopt(long = "rpc", default_value = "127.0.0.1:11055")]
-    pub rpc_listen: SocketAddr,
+    #[structopt(long = "rpc", default_value = "tcp://127.0.0.1:11055")]
+    pub rpc_listen: String,
     /// Sets Datastore Path
     #[structopt(long, default_value = "~/.config/tau")]
     pub datastore: String,

src/rpc/mod.rs

@@ -1,5 +1,3 @@
 pub mod jsonrpc;
 pub mod rpcserver;
-// TODO: Replace rpcserver with this
-pub mod rpcserver2;
 pub mod websockets;

src/rpc/rpcserver.rs

@@ -1,164 +1,180 @@
-use std::{
-    net::{SocketAddr, TcpListener, TcpStream},
-    path::PathBuf,
+use async_std::{
+    io::{ReadExt, WriteExt},
     sync::Arc,
 };
+use std::{env, fs};
 
-use async_executor::Executor;
-use async_native_tls::{Identity, TlsAcceptor};
 use async_trait::async_trait;
 use log::{debug, error, info};
-use smol::{
-    io::{AsyncReadExt, AsyncWriteExt},
-    Async,
-};
+use url::Url;
 
+use super::jsonrpc::{JsonRequest, JsonResult};
 use crate::{
-    rpc::jsonrpc::{JsonRequest, JsonResult},
-    Result,
+    net::transport::{
+        TcpTransport, TorTransport, Transport, TransportListener, TransportName, TransportStream,
+    },
+    Error, Result,
 };
 
-pub struct RpcServerConfig {
-    pub socket_addr: SocketAddr,
-    pub use_tls: bool,
-    pub identity_path: PathBuf,
-    pub identity_pass: String,
-}
-
 #[async_trait]
 pub trait RequestHandler: Sync + Send {
-    async fn handle_request(&self, req: JsonRequest, executor: Arc<Executor<'_>>) -> JsonResult;
+    async fn handle_request(&self, req: JsonRequest) -> JsonResult;
 }
 
-async fn serve(
-    mut stream: Async<TcpStream>,
-    tls: Option<TlsAcceptor>,
+async fn run_accept_loop(
+    listener: Box<dyn TransportListener>,
     rh: Arc<impl RequestHandler + 'static>,
-    executor: Arc<Executor<'_>>,
 ) -> Result<()> {
-    debug!(target: "RPC SERVER", "Accepted connection");
-
-    let mut buf = [0; 8192];
+    // TODO can we spawn new task here ?
+    while let Ok((stream, peer_addr)) = listener.next().await {
+        info!(target: "JSON-RPC SERVER", "RPC Accepted connection {}", peer_addr);
+        accept(stream, rh.clone()).await?;
+    }
+    Ok(())
+}
 
-    match tls {
-        None => loop {
-            let n = match stream.read(&mut buf).await {
-                Ok(n) if n == 0 => {
-                    debug!(target: "RPC SERVER", "Closed connection");
-                    return Ok(())
-                }
-                Ok(n) => n,
-                Err(e) => {
-                    debug!(target: "RPC SERVER", "Failed to read from socket: {:#?}", e);
-                    debug!(target: "RPC SERVER", "Closed connection");
-                    return Ok(())
-                }
-            };
-
-            let r: JsonRequest = match serde_json::from_slice(&buf[0..n]) {
-                Ok(r) => r,
-                Err(e) => {
-                    debug!(target: "RPC SERVER", "Received invalid JSON: {:#?}", e);
-                    debug!(target: "RPC SERVER", "Closed connection");
-                    return Ok(())
-                }
-            };
+async fn accept(
+    mut stream: Box<dyn TransportStream>,
+    rh: Arc<impl RequestHandler + 'static>,
+) -> Result<()> {
+    let mut buf = vec![0; 8192];
 
-            let reply = rh.handle_request(r, executor.clone()).await;
-            let j = serde_json::to_string(&reply)?;
-            debug!(target: "RPC", "<-- {}", j);
+    loop {
+        let n = match stream.read(&mut buf).await {
+            Ok(n) if n == 0 => {
+                info!(target: "JSON-RPC SERVER", "Closed connection");
+                break
+            }
+            Ok(n) => n,
+            Err(e) => {
+                error!(target: "JSON-RPC SERVER", "Failed reading from socket: {}", e);
+                info!(target: "JSON-RPC SERVER", "Closed connection");
+                break
+            }
+        };
 
-            if let Err(e) = stream.write_all(j.as_bytes()).await {
-                debug!(target: "RPC SERVER", "Failed to write to socket: {:#?}", e);
-                debug!(target: "RPC SERVER", "Closed connection");
-                return Ok(())
+        let r: JsonRequest = match serde_json::from_slice(&buf[0..n]) {
+            Ok(r) => {
+                debug!(target: "JSON-RPC SERVER", "--> {}", String::from_utf8_lossy(&buf));
+                r
             }
-        },
-        Some(tls) => match tls.accept(stream).await {
-            Ok(mut stream) => loop {
-                let n = match stream.read(&mut buf).await {
-                    Ok(n) if n == 0 => {
-                        debug!(target: "RPC SERVER", "Closed connection");
-                        return Ok(())
-                    }
-                    Ok(n) => n,
-                    Err(e) => {
-                        debug!(target: "RPC SERVER", "Failed to read from socket: {:#?}", e);
-                        debug!(target: "RPC SERVER", "Closed connection");
-                        return Ok(())
-                    }
-                };
-
-                let r: JsonRequest = match serde_json::from_slice(&buf[0..n]) {
-                    Ok(r) => r,
-                    Err(e) => {
-                        debug!(target: "RPC SERVER", "Received invalid JSON: {:#?}", e);
-                        debug!(target: "RPC SERVER", "Closed connection");
-                        return Ok(())
-                    }
-                };
-
-                let reply = rh.handle_request(r, executor.clone()).await;
-                let j = serde_json::to_string(&reply)?;
-                debug!(target: "RPC", "<-- {}", j);
-
-                if let Err(e) = stream.write_all(j.as_bytes()).await {
-                    debug!(target: "RPC SERVER", "Failed to write to socket: {:#?}", e);
-                    return Ok(())
-                }
-            },
             Err(e) => {
-                debug!(target: "RPC SERVER", "Failed to establish TLS connection: {:#}", e);
-                Ok(())
+                error!(target: "JSON-RPC SERVER", "Received invalid JSON: {:?}", e);
+                info!(target: "JSON-RPC SERVER", "Closed connection");
+                break
             }
-        },
+        };
+
+        let reply = rh.handle_request(r).await;
+        let j = serde_json::to_string(&reply)?;
+        debug!(target: "JSON-RPC SERVER", "<-- {}", j);
+
+        if let Err(e) = stream.write_all(j.as_bytes()).await {
+            error!(target: "JSON-RPC SERVER", "Failed writing to socket: {}", e);
+            info!(target: "JSON-RPC SERVER", "Closed connection");
+            break
+        }
     }
+
+    Ok(())
 }
 
-async fn listen(
-    listener: Async<TcpListener>,
-    tls: Option<TlsAcceptor>,
+pub async fn listen_and_serve(
+    accept_url: Url,
     rh: Arc<impl RequestHandler + 'static>,
-    executor: Arc<Executor<'_>>,
 ) -> Result<()> {
-    match &tls {
-        None => {
-            info!(target: "RPC SERVER", "Listening on tcp://{}", listener.get_ref().local_addr()?)
-        }
-        Some(_) => {
-            info!(target: "RPC SERVER", "Listening on tls://{}", listener.get_ref().local_addr()?)
+    debug!(target: "JSON-RPC SERVER", "Trying to start listener on {}", accept_url);
+
+    let transport_name = TransportName::try_from(accept_url.clone())?;
+    match transport_name {
+        TransportName::Tcp(upgrade) => {
+            let transport = TcpTransport::new(None, 1024);
+            let listener = transport.listen_on(accept_url.clone());
+
+            if let Err(err) = listener {
+                error!("TCP Setup failed: {}", err);
+                return Err(Error::BindFailed(accept_url.clone().to_string()))
+            }
+
+            let listener = listener?.await;
+
+            if let Err(err) = listener {
+                error!("TCP Bind listener failed: {}", err);
+                return Err(Error::BindFailed(accept_url.to_string()))
+            }
+
+            let listener = listener?;
+
+            match upgrade {
+                None => {
+                    run_accept_loop(Box::new(listener), rh).await?;
+                }
+                Some(u) if u == "tls" => {
+                    let tls_listener = transport.upgrade_listener(listener)?.await?;
+                    run_accept_loop(Box::new(tls_listener), rh).await?;
+                }
+                Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
+            }
         }
-    }
+        TransportName::Tor(upgrade) => {
+            let socks5_url = Url::parse(
+                &env::var("DARKFI_TOR_SOCKS5_URL").unwrap_or("socks5://127.0.0.1:9050".to_string()),
+            )?;
+
+            let torc_url = Url::parse(
+                &env::var("DARKFI_TOR_CONTROL_URL").unwrap_or("tcp://127.0.0.1:9051".to_string()),
+            )?;
+
+            let auth_cookie = env::var("DARKFI_TOR_COOKIE");
+
+            if auth_cookie.is_err() {
+                return Err(Error::TorError(
+                    "Please set the env var DARKFI_TOR_COOKIE to the configured tor cookie file. \
+                    For example: \
+                    \'export DARKFI_TOR_COOKIE=\"/var/lib/tor/control_auth_cookie\"\'"
+                        .to_string(),
+                ))
+            }
 
-    let ex = executor.clone();
-    loop {
-        let (stream, _) = listener.accept().await?;
-        let tls = tls.clone();
-        let rh_c = rh.clone();
-
-        let ex2 = ex.clone();
-        ex.spawn(async move {
-            if let Err(err) = serve(stream, tls, rh_c, ex2.clone()).await {
-                error!(target: "RPC SERVER", "Connection error: {:#?}", err);
+            let auth_cookie = auth_cookie.unwrap();
+
+            let auth_cookie = hex::encode(&fs::read(auth_cookie).unwrap());
+
+            let transport = TorTransport::new(socks5_url, Some((torc_url, auth_cookie)))?;
+
+            // generate EHS pointing to local address
+            let hurl = transport.create_ehs(accept_url.clone())?;
+
+            info!("EHS TOR: {}", hurl.to_string());
+
+            let listener = transport.clone().listen_on(accept_url.clone());
+
+            if let Err(err) = listener {
+                error!("TOR Setup failed: {}", err);
+                return Err(Error::BindFailed(accept_url.clone().to_string()))
             }
-        })
-        .detach();
+
+            let listener = listener?.await;
+
+            if let Err(err) = listener {
+                error!("TOR Bind listener failed: {}", err);
+                return Err(Error::BindFailed(accept_url.to_string()))
+            }
+
+            let listener = listener?;
+
+            match upgrade {
+                None => {
+                    run_accept_loop(Box::new(listener), rh).await?;
+                }
+                Some(u) if u == "tls" => {
+                    let tls_listener = transport.upgrade_listener(listener)?.await?;
+                    run_accept_loop(Box::new(tls_listener), rh).await?;
+                }
+                Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
+            }
+        }
     }
-}
 
-pub async fn listen_and_serve(
-    cfg: RpcServerConfig,
-    rh: Arc<impl RequestHandler + 'static>,
-    executor: Arc<Executor<'_>>,
-) -> Result<()> {
-    let tls: Option<TlsAcceptor> = if cfg.use_tls {
-        let ident_bytes = std::fs::read(cfg.identity_path)?;
-        let identity = Identity::from_pkcs12(&ident_bytes, &cfg.identity_pass)?;
-        Some(TlsAcceptor::from(native_tls::TlsAcceptor::new(identity)?))
-    } else {
-        None
-    };
-
-    let listener = listen(Async::<TcpListener>::bind(cfg.socket_addr)?, tls, rh, executor);
-    listener.await
+    Ok(())
 }

src/rpc/rpcserver2.rs

@@ -1,180 +0,0 @@
-use async_std::{
-    io::{ReadExt, WriteExt},
-    sync::Arc,
-};
-use std::{env, fs};
-
-use async_trait::async_trait;
-use log::{debug, error, info};
-use url::Url;
-
-use super::jsonrpc::{JsonRequest, JsonResult};
-use crate::{
-    net::transport::{
-        TcpTransport, TorTransport, Transport, TransportListener, TransportName, TransportStream,
-    },
-    Error, Result,
-};
-
-#[async_trait]
-pub trait RequestHandler: Sync + Send {
-    async fn handle_request(&self, req: JsonRequest) -> JsonResult;
-}
-
-async fn run_accept_loop(
-    listener: Box<dyn TransportListener>,
-    rh: Arc<impl RequestHandler + 'static>,
-) -> Result<()> {
-    // TODO can we spawn new task here ?
-    while let Ok((stream, peer_addr)) = listener.next().await {
-        info!(target: "JSON-RPC SERVER", "RPC Accepted connection {}", peer_addr);
-        accept(stream, rh.clone()).await?;
-    }
-    Ok(())
-}
-
-async fn accept(
-    mut stream: Box<dyn TransportStream>,
-    rh: Arc<impl RequestHandler + 'static>,
-) -> Result<()> {
-    let mut buf = vec![0; 8192];
-
-    loop {
-        let n = match stream.read(&mut buf).await {
-            Ok(n) if n == 0 => {
-                info!(target: "JSON-RPC SERVER", "Closed connection");
-                break
-            }
-            Ok(n) => n,
-            Err(e) => {
-                error!(target: "JSON-RPC SERVER", "Failed reading from socket: {}", e);
-                info!(target: "JSON-RPC SERVER", "Closed connection");
-                break
-            }
-        };
-
-        let r: JsonRequest = match serde_json::from_slice(&buf[0..n]) {
-            Ok(r) => {
-                debug!(target: "JSON-RPC SERVER", "--> {}", String::from_utf8_lossy(&buf));
-                r
-            }
-            Err(e) => {
-                error!(target: "JSON-RPC SERVER", "Received invalid JSON: {:?}", e);
-                info!(target: "JSON-RPC SERVER", "Closed connection");
-                break
-            }
-        };
-
-        let reply = rh.handle_request(r).await;
-        let j = serde_json::to_string(&reply)?;
-        debug!(target: "JSON-RPC SERVER", "<-- {}", j);
-
-        if let Err(e) = stream.write_all(j.as_bytes()).await {
-            error!(target: "JSON-RPC SERVER", "Failed writing to socket: {}", e);
-            info!(target: "JSON-RPC SERVER", "Closed connection");
-            break
-        }
-    }
-
-    Ok(())
-}
-
-pub async fn listen_and_serve(
-    accept_url: Url,
-    rh: Arc<impl RequestHandler + 'static>,
-) -> Result<()> {
-    debug!(target: "JSON-RPC SERVER", "Trying to start listener on {}", accept_url);
-
-    let transport_name = TransportName::try_from(accept_url.clone())?;
-    match transport_name {
-        TransportName::Tcp(upgrade) => {
-            let transport = TcpTransport::new(None, 1024);
-            let listener = transport.listen_on(accept_url.clone());
-
-            if let Err(err) = listener {
-                error!("TCP Setup failed: {}", err);
-                return Err(Error::BindFailed(accept_url.clone().to_string()))
-            }
-
-            let listener = listener?.await;
-
-            if let Err(err) = listener {
-                error!("TCP Bind listener failed: {}", err);
-                return Err(Error::BindFailed(accept_url.to_string()))
-            }
-
-            let listener = listener?;
-
-            match upgrade {
-                None => {
-                    run_accept_loop(Box::new(listener), rh).await?;
-                }
-                Some(u) if u == "tls" => {
-                    let tls_listener = transport.upgrade_listener(listener)?.await?;
-                    run_accept_loop(Box::new(tls_listener), rh).await?;
-                }
-                Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
-            }
-        }
-        TransportName::Tor(upgrade) => {
-            let socks5_url = Url::parse(
-                &env::var("DARKFI_TOR_SOCKS5_URL").unwrap_or("socks5://127.0.0.1:9050".to_string()),
-            )?;
-
-            let torc_url = Url::parse(
-                &env::var("DARKFI_TOR_CONTROL_URL").unwrap_or("tcp://127.0.0.1:9051".to_string()),
-            )?;
-
-            let auth_cookie = env::var("DARKFI_TOR_COOKIE");
-
-            if auth_cookie.is_err() {
-                return Err(Error::TorError(
-                    "Please set the env var DARKFI_TOR_COOKIE to the configured tor cookie file. \
-                    For example: \
-                    \'export DARKFI_TOR_COOKIE=\"/var/lib/tor/control_auth_cookie\"\'"
-                        .to_string(),
-                ))
-            }
-
-            let auth_cookie = auth_cookie.unwrap();
-
-            let auth_cookie = hex::encode(&fs::read(auth_cookie).unwrap());
-
-            let transport = TorTransport::new(socks5_url, Some((torc_url, auth_cookie)))?;
-
-            // generate EHS pointing to local address
-            let hurl = transport.create_ehs(accept_url.clone())?;
-
-            info!("EHS TOR: {}", hurl.to_string());
-
-            let listener = transport.clone().listen_on(accept_url.clone());
-
-            if let Err(err) = listener {
-                error!("TOR Setup failed: {}", err);
-                return Err(Error::BindFailed(accept_url.clone().to_string()))
-            }
-
-            let listener = listener?.await;
-
-            if let Err(err) = listener {
-                error!("TOR Bind listener failed: {}", err);
-                return Err(Error::BindFailed(accept_url.to_string()))
-            }
-
-            let listener = listener?;
-
-            match upgrade {
-                None => {
-                    run_accept_loop(Box::new(listener), rh).await?;
-                }
-                Some(u) if u == "tls" => {
-                    let tls_listener = transport.upgrade_listener(listener)?.await?;
-                    run_accept_loop(Box::new(tls_listener), rh).await?;
-                }
-                Some(u) => return Err(Error::UnsupportedTransportUpgrade(u)),
-            }
-        }
-    }
-
-    Ok(())
-}