Functional seed synchronization step

scripts/jsonrpc_client.py

@@ -7,7 +7,8 @@ def main():
 
     # Example echo method
     payload = {
-        "method": "get_info",
+        "method": "stop",
+        #"method": "get_info",
         "params": [],
         "jsonrpc": "2.0",
         "id": 0,

src/bin/dfi.rs

@@ -75,19 +75,19 @@ async fn listen(
 struct RpcInterface {
     p2p: Arc<net::P2p>,
     started: Mutex<bool>,
-    quit_send: async_channel::Sender<()>,
-    quit_recv: async_channel::Receiver<()>,
+    stop_send: async_channel::Sender<()>,
+    stop_recv: async_channel::Receiver<()>,
 }
 
 impl RpcInterface {
     fn new(p2p: Arc<net::P2p>) -> Arc<Self> {
-        let (quit_send, quit_recv) = async_channel::unbounded::<()>();
+        let (stop_send, stop_recv) = async_channel::unbounded::<()>();
 
         Arc::new(Self {
             p2p,
             started: Mutex::new(false),
-            quit_send,
-            quit_recv,
+            stop_send,
+            stop_recv,
         })
     }
 
@@ -104,14 +104,19 @@ impl RpcInterface {
         let self2 = self.clone();
         io.add_method("get_info", move |_| {
             let self2 = self2.clone();
-            async move { Ok(json!({"started": *self2.started.lock().await})) }
+            async move {
+                Ok(json!({
+                    "started": *self2.started.lock().await,
+                    "connections": self2.p2p.connections_count().await
+                }))
+            }
         });
 
-        let quit_send = self.quit_send.clone();
-        io.add_method("quit", move |_| {
-            let quit_send = quit_send.clone();
+        let stop_send = self.stop_send.clone();
+        io.add_method("stop", move |_| {
+            let stop_send = stop_send.clone();
             async move {
-                let _ = quit_send.send(()).await;
+                let _ = stop_send.send(()).await;
                 Ok(jsonrpc_core::Value::Null)
             }
         });
@@ -127,7 +132,7 @@ impl RpcInterface {
     }
 
     async fn wait_for_quit(self: Arc<Self>) -> Result<()> {
-        Ok(self.quit_recv.recv().await?)
+        Ok(self.stop_recv.recv().await?)
     }
 }
 
@@ -138,7 +143,7 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
     let http = listen(
         executor.clone(),
         rpc.clone(),
-        Async::<TcpListener>::bind(([127, 0, 0, 1], 8000))?,
+        Async::<TcpListener>::bind(([127, 0, 0, 1], options.rpc_port))?,
         None,
     );
 
@@ -146,7 +151,8 @@ async fn start(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<(
 
     *rpc.started.lock().await = true;
 
-    p2p.start(executor.clone()).await?;
+    p2p.clone().start(executor.clone()).await?;
+    p2p.run(executor).await?;
 
     rpc.wait_for_quit().await?;
 
@@ -232,7 +238,7 @@ async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<
 
     let http_task = executor.spawn(http);
 
-    rpc.quit_recv.recv().await?;
+    rpc.stop_recv.recv().await?;
 
     http_task.cancel().await;
 
@@ -249,6 +255,7 @@ async fn start2(executor: Arc<Executor<'_>>, options: ProgramOptions) -> Result<
 struct ProgramOptions {
     network_settings: net::Settings,
     log_path: Box<std::path::PathBuf>,
+    rpc_port: u16,
 }
 
 impl ProgramOptions {
@@ -262,6 +269,8 @@ impl ProgramOptions {
             (@arg CONNECTS: -c --connect ... "Manual connections")
             (@arg CONNECT_SLOTS: --slots +takes_value "Connection slots")
             (@arg LOG_PATH: --log +takes_value "Logfile path")
+            (@arg DISABLE_SEED: -D --disable_seed "Disable seed process")
+            (@arg RPC_PORT: -r --rpc +takes_value "RPC port")
         )
         .get_matches();
 
@@ -300,6 +309,14 @@ impl ProgramOptions {
             .to_path_buf(),
         );
 
+        let skip_seed_sync = if app.is_present("DISABLE_SEED") { true } else { false };
+
+        let rpc_port = if let Some(rpc_port) = app.value_of("RPC_PORT") {
+            rpc_port.parse()?
+        } else {
+            8000
+        };
+
         Ok(ProgramOptions {
             network_settings: net::Settings {
                 inbound: accept_addr,
@@ -310,8 +327,10 @@ impl ProgramOptions {
                 external_addr: accept_addr,
                 peers: manual_connects,
                 seeds: seed_addrs,
+                skip_seed_sync
             },
             log_path,
+            rpc_port,
         })
     }
 }

src/net/channel.rs

@@ -97,7 +97,7 @@ impl Channel {
                     match message_result {
                         Ok(message) => Ok(Arc::new(message)),
                         Err(err) => {
-                            error!("Read error on channel {}", err);
+                            error!("Read error on channel: {}", err);
                             self.stop().await;
                             Err(NetError::ChannelStopped)
                         }

src/net/message_subscriber.rs

@@ -71,7 +71,7 @@ impl MessageSubscription {
 }
 
 pub struct MessageSubscriber {
-    subs: Mutex<HashMap<u64, async_channel::Sender<MessageResult>>>,
+    subs: Mutex<HashMap<MessageSubscriptionID, async_channel::Sender<MessageResult>>>,
 }
 
 impl MessageSubscriber {
@@ -106,13 +106,26 @@ impl MessageSubscriber {
     }
 
     pub async fn notify(&self, message_result: NetResult<Arc<Message>>) {
-        for sub in (*self.subs.lock().await).values() {
+        let mut garbage_ids = Vec::new();
+
+        for (sub_id, sub) in &*self.subs.lock().await {
             match sub.send(message_result.clone()).await {
                 Ok(()) => {}
-                Err(err) => {
-                    panic!("Error returned sending message in notify() call! {}", err);
+                Err(_err) => {
+                    // Automatically clean out closed channels
+                    garbage_ids.push(*sub_id);
+                    //panic!("Error returned sending message in notify() call! {}", err);
                 }
             }
         }
+
+        self.collect_garbage(garbage_ids).await;
+    }
+
+    async fn collect_garbage(&self, ids: Vec<MessageSubscriptionID>) {
+        let mut subs = self.subs.lock().await;
+        for id in &ids {
+            subs.remove(id);
+        }
     }
 }

src/net/p2p.rs

@@ -31,9 +31,11 @@ impl P2p {
     /// Invoke startup and seeding sequence. Call from constructing thread.
     pub async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> NetResult<()> {
         // Start manual connections
+
         // Start seed session
         let seed = SeedSession::new(Arc::downgrade(&self));
         seed.start(executor.clone()).await?;
+
         Ok(())
     }
 
@@ -58,6 +60,10 @@ impl P2p {
             .remove(&channel.address());
     }
 
+    pub async fn connections_count(&self) -> usize {
+        self.pending_channels.lock().await.len()
+    }
+
     pub fn settings(&self) -> SettingsPtr {
         self.settings.clone()
     }

src/net/sessions/seed_session.rs

@@ -23,6 +23,11 @@ impl SeedSession {
             p2p.settings()
         };
 
+        if settings.skip_seed_sync {
+            info!("Configured to skip seed synchronization process.");
+            return Ok(());
+        }
+
         // if cached addresses then quit
 
         // if seeds empty then seeding required but empty

src/net/settings.rs

@@ -15,4 +15,5 @@ pub struct Settings {
     pub external_addr: Option<SocketAddr>,
     pub peers: Vec<SocketAddr>,
     pub seeds: Vec<SocketAddr>,
+    pub skip_seed_sync: bool,
 }