feat: Integrate discovery and identify protocol

network/Cargo.toml

@@ -20,9 +20,11 @@ bytes = "0.4.12"
 tokio = "0.1.17"
 futures = { version = "0.1.19", features = ["use_std"] }
 snap = "0.2"
-p2p = { version = "0.1", package="tentacle" }
-p2p-ping = { version = "0.1", package="tentacle-ping" }
-secio = { version = "0.1", package="tentacle-secio" }
+p2p = { git = "https://github.com/nervosnetwork/p2p", rev="9c42be05724c3e2decb73716d0f515c36a7af1cb", package="tentacle" }
+secio = { git = "https://github.com/nervosnetwork/p2p", rev="9c42be05724c3e2decb73716d0f515c36a7af1cb", package="tentacle-secio" }
+p2p-ping = { git = "https://github.com/nervosnetwork/p2p", rev="9c42be05724c3e2decb73716d0f515c36a7af1cb", package="tentacle-ping" }
+p2p-discovery = { git = "https://github.com/nervosnetwork/p2p", rev="9c42be05724c3e2decb73716d0f515c36a7af1cb", package="tentacle-discovery" }
+p2p-identify = { git = "https://github.com/nervosnetwork/p2p", rev="9c42be05724c3e2decb73716d0f515c36a7af1cb", package="tentacle-identify" }
 faketime = "0.2.0"
 rusqlite = {version = "0.16.0", features = ["bundled"]}
 lazy_static = "1.3.0"

network/src/network.rs

@@ -7,6 +7,8 @@ use crate::protocol::Version as ProtocolVersion;
 use crate::protocol_handler::{CKBProtocolHandler, DefaultCKBProtocolContext};
 use crate::service::{
     ckb_service::CKBService,
+    discovery_service::{DiscoveryEvent, DiscoveryProtocol, DiscoveryService},
+    identify_service::{IdentifyAddressManager, IdentifyEvent, IdentifyProtocol, IdentifyService},
     outbound_peer_service::OutboundPeerService,
     ping_service::PingService,
     timer_service::{TimerRegistry, TimerService},
@@ -19,29 +21,31 @@ use bytes::Bytes;
 use ckb_util::{Mutex, RwLock};
 use fnv::FnvHashMap;
 use futures::future::{select_all, Future};
+use futures::sync::mpsc;
 use futures::sync::mpsc::channel;
 use futures::sync::mpsc::Receiver;
 use futures::sync::oneshot;
 use futures::Stream;
 use log::{debug, error, info, warn};
 use multiaddr::multihash::Multihash;
 use p2p::{
-    builder::ServiceBuilder,
+    builder::{MetaBuilder, ServiceBuilder},
     multiaddr::{self, Multiaddr},
     secio::{PeerId, PublicKey},
-    service::{Service, ServiceError, ServiceEvent},
+    service::{DialProtocol, ProtocolHandle, Service, ServiceError, ServiceEvent},
     traits::ServiceHandle,
 };
-use p2p_ping::{Event as PingEvent, PingProtocol};
+use p2p_ping::{Event as PingEvent, PingHandler};
 use secio;
 use std::boxed::Box;
 use std::cmp::max;
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use std::usize;
-use tokio::codec::LengthDelimitedCodec;
 
 const PING_PROTOCOL_ID: ProtocolId = 0;
+const DISCOVERY_PROTOCOL_ID: ProtocolId = 1;
+const IDENTIFY_PROTOCOL_ID: ProtocolId = 2;
 
 pub type CKBProtocols = Vec<(CKBProtocol, Arc<dyn CKBProtocolHandler>)>;
 type NetworkResult = Result<
@@ -80,7 +84,13 @@ impl PeerInfo {
     }
 }
 
-type P2PService = Service<EventHandler, LengthDelimitedCodec>;
+pub struct EventReceivers {
+    pub ping_event_receiver: Receiver<PingEvent>,
+    pub disc_event_receiver: mpsc::UnboundedReceiver<DiscoveryEvent>,
+    pub identify_event_receiver: mpsc::UnboundedReceiver<IdentifyEvent>,
+}
+
+type P2PService = Service<EventHandler>;
 
 pub struct Network {
     pub(crate) peers_registry: RwLock<PeersRegistry>,
@@ -313,7 +323,7 @@ impl Network {
     }
 
     pub fn dial_addr(&self, addr: Multiaddr) {
-        if let Err(err) = self.p2p_control.write().dial(addr) {
+        if let Err(err) = self.p2p_control.write().dial(addr, DialProtocol::All) {
             error!(target: "network", "failed to dial: {}", err);
         }
     }
@@ -338,7 +348,7 @@ impl Network {
     pub(crate) fn inner_build(
         config: &NetworkConfig,
         ckb_protocols: CKBProtocols,
-    ) -> Result<(Arc<Self>, P2PService, TimerRegistry, Receiver<PingEvent>), Error> {
+    ) -> Result<(Arc<Self>, P2PService, TimerRegistry, EventReceivers), Error> {
         let local_private_key = match config.fetch_private_key() {
             Some(private_key) => private_key?,
             None => return Err(ConfigError::InvalidKey.into()),
@@ -369,17 +379,50 @@ impl Network {
             config.reserved_only,
             reserved_peers,
         );
-        let mut p2p_service = ServiceBuilder::default().forever(true);
         // register protocols
         let (ping_sender, ping_receiver) = channel(std::u8::MAX as usize);
-        p2p_service = p2p_service.insert_protocol(PingProtocol::new(
-            PING_PROTOCOL_ID,
-            config.ping_interval,
-            config.ping_timeout,
-            ping_sender,
-        ));
+        let ping_meta = MetaBuilder::default()
+            .id(PING_PROTOCOL_ID)
+            .service_handle(move || {
+                ProtocolHandle::Callback(Box::new(PingHandler::new(
+                    PING_PROTOCOL_ID,
+                    config.ping_interval,
+                    config.ping_timeout,
+                    ping_sender,
+                )))
+            })
+            .build();
+
+        let (disc_sender, disc_receiver) = mpsc::unbounded();
+        let disc_meta = MetaBuilder::default()
+            .id(DISCOVERY_PROTOCOL_ID)
+            .service_handle(move || {
+                ProtocolHandle::Callback(Box::new(DiscoveryProtocol::new(
+                    DISCOVERY_PROTOCOL_ID,
+                    disc_sender,
+                )))
+            })
+            .build();
+
+        let (identify_sender, identify_receiver) = mpsc::unbounded();
+        let identify_addr_mgr = IdentifyAddressManager::new(identify_sender.clone());
+        let identify_meta = MetaBuilder::default()
+            .id(IDENTIFY_PROTOCOL_ID)
+            .service_handle(move || {
+                ProtocolHandle::Callback(Box::new(IdentifyProtocol::new(
+                    IDENTIFY_PROTOCOL_ID,
+                    identify_addr_mgr.clone(),
+                )))
+            })
+            .build();
+
+        let mut p2p_service = ServiceBuilder::default()
+            .forever(true)
+            .insert_protocol(ping_meta)
+            .insert_protocol(disc_meta)
+            .insert_protocol(identify_meta);
         for (ckb_protocol, _) in &ckb_protocols {
-            p2p_service = p2p_service.insert_protocol(ckb_protocol.clone());
+            p2p_service = p2p_service.insert_protocol(ckb_protocol.build());
         }
         let mut p2p_service = p2p_service
             .key_pair(local_private_key.clone())
@@ -453,7 +496,16 @@ impl Network {
             }
         }
 
-        Ok((network, p2p_service, timer_registry, ping_receiver))
+        Ok((
+            network,
+            p2p_service,
+            timer_registry,
+            EventReceivers {
+                ping_event_receiver: ping_receiver,
+                disc_event_receiver: disc_receiver,
+                identify_event_receiver: identify_receiver,
+            },
+        ))
     }
 
     pub(crate) fn build_network_future(
@@ -463,23 +515,21 @@ impl Network {
         p2p_service: P2PService,
         timer_registry: TimerRegistry,
         ckb_event_receiver: Receiver<CKBEvent>,
-        ping_event_receiver: Receiver<PingEvent>,
+        receivers: EventReceivers,
     ) -> Result<Box<Future<Item = (), Error = Error> + Send>, Error> {
         // initialize ckb_protocols
         let ping_service = PingService {
             network: Arc::clone(&network),
-            event_receiver: ping_event_receiver,
+            event_receiver: receivers.ping_event_receiver,
         };
-        //let identify_service = Arc::new(IdentifyService {
-        //    client_version,
-        //    protocol_version,
-        //    identify_timeout: config.identify_timeout,
-        //    identify_interval: config.identify_interval,
-        //});
+        let disc_service =
+            DiscoveryService::new(Arc::clone(&network), receivers.disc_event_receiver);
+        let identify_service =
+            IdentifyService::new(Arc::clone(&network), receivers.identify_event_receiver);
 
         let ckb_service = CKBService {
-            event_receiver: ckb_event_receiver,
             network: Arc::clone(&network),
+            event_receiver: ckb_event_receiver,
         };
         let timer_service = TimerService::new(timer_registry, Arc::clone(&network));
         let outbound_peer_service =
@@ -501,17 +551,16 @@ impl Network {
                     .for_each(|_| Ok(()))
                     .map_err(|_err| Error::Shutdown),
             ),
-            // Box::new(
-            //     discovery_query_service
-            //         .into_future()
-            //         .map(|_| ())
-            //         .map_err(|(err, _)| err),
-            // ) as Box<Future<Item = (), Error = IoError> + Send>,
-            //identify_service.start_protocol(
-            //    Arc::clone(&network),
-            //    swarm_controller.clone(),
-            //    basic_transport.clone(),
-            //),
+            Box::new(
+                disc_service
+                    .for_each(|_| Ok(()))
+                    .map_err(|_err| Error::Shutdown),
+            ),
+            Box::new(
+                identify_service
+                    .for_each(|_| Ok(()))
+                    .map_err(|_err| Error::Shutdown),
+            ),
             Box::new(timer_service.timer_futures.for_each(|_| Ok(()))),
             Box::new(
                 outbound_peer_service
@@ -543,7 +592,7 @@ impl Network {
         ckb_protocols: CKBProtocols,
         ckb_event_receiver: Receiver<CKBEvent>,
     ) -> NetworkResult {
-        let (network, p2p_service, timer_registry, ping_event_receiver) =
+        let (network, p2p_service, timer_registry, receivers) =
             Self::inner_build(config, ckb_protocols)?;
         let (close_tx, close_rx) = oneshot::channel();
         let network_future = Self::build_network_future(
@@ -553,7 +602,7 @@ impl Network {
             p2p_service,
             timer_registry,
             ckb_event_receiver,
-            ping_event_receiver,
+            receivers,
         )?;
         Ok((network, close_tx, network_future))
     }

network/src/network_service.rs

@@ -9,27 +9,34 @@ use ckb_util::Mutex;
 use futures::future::Future;
 use futures::sync::mpsc::Receiver;
 use futures::sync::oneshot;
-use log::{debug, info};
+use log::{debug, error, info};
 use std::sync::Arc;
-use std::thread;
 use tokio::runtime;
 
 pub struct StopHandler {
     signal: oneshot::Sender<()>,
-    thread: thread::JoinHandle<()>,
+    network_runtime: runtime::Runtime,
 }
 
 impl StopHandler {
-    pub fn new(signal: oneshot::Sender<()>, thread: thread::JoinHandle<()>) -> StopHandler {
-        StopHandler { signal, thread }
+    pub fn new(signal: oneshot::Sender<()>, network_runtime: runtime::Runtime) -> StopHandler {
+        StopHandler {
+            signal,
+            network_runtime,
+        }
     }
 
     pub fn close(self) {
-        let StopHandler { signal, thread } = self;
+        let StopHandler {
+            signal,
+            network_runtime,
+        } = self;
         if let Err(e) = signal.send(()) {
             debug!(target: "network", "send shutdown signal error, ignoring error: {:?}", e)
         };
-        thread.join().expect("join network_service thread");
+        // TODO: not that gracefully shutdown, will output below error message:
+        //       "terminate called after throwing an instance of 'std::system_error'"
+        network_runtime.shutdown_now();
     }
 }
 
@@ -67,42 +74,39 @@ impl NetworkService {
         ckb_protocols: CKBProtocols,
         ckb_event_receiver: Receiver<CKBEvent>,
     ) -> Result<NetworkService, Error> {
-        let (network, p2p_service, timer_registry, ping_event_receiver) =
+        let (network, p2p_service, timer_registry, receivers) =
             Network::inner_build(config, ckb_protocols)?;
         let (close_tx, close_rx) = oneshot::channel();
         let (init_tx, init_rx) = oneshot::channel();
-        let join_handle = thread::spawn({
-            let network = Arc::clone(&network);
-            let config = config.clone();
-            move || {
-                info!(
-                    target: "network",
-                    "network peer_id {:?}",
-                    network.local_public_key().peer_id()
-                );
-                let network_future = Network::build_network_future(
-                    network,
-                    &config,
-                    close_rx,
-                    p2p_service,
-                    timer_registry,
-                    ckb_event_receiver,
-                    ping_event_receiver,
-                )
-                .expect("Network thread init");
-                init_tx.send(()).expect("Network init signal send");
-                // here we use default config
-                let network_runtime = runtime::Runtime::new().expect("Network tokio runtime init");;
-                match network_runtime.block_on_all(network_future) {
-                    Ok(_) => info!(target: "network", "network service exit"),
-                    Err(err) => panic!("network service exit unexpected {}", err),
-                }
-            }
-        });
+
+        info!(
+            target: "network",
+            "network peer_id {:?}",
+            network.local_public_key().peer_id()
+        );
+        let network_future = Network::build_network_future(
+            Arc::clone(&network),
+            &config,
+            close_rx,
+            p2p_service,
+            timer_registry,
+            ckb_event_receiver,
+            receivers,
+        )
+        .expect("Network thread init");
+        init_tx.send(()).expect("Network init signal send");
+        // here we use default config
+        let mut network_runtime = runtime::Runtime::new().expect("Network tokio runtime init");
+        network_runtime.spawn(
+            network_future
+                .map(|_| info!(target: "network", "network service exit"))
+                .map_err(|err| error!("network service exit unexpected {}", err)),
+        );
+
         init_rx.wait().map_err(|_err| Error::Shutdown)?;
         Ok(NetworkService {
             network,
-            stop_handler: Mutex::new(Some(StopHandler::new(close_tx, join_handle))),
+            stop_handler: Mutex::new(Some(StopHandler::new(close_tx, network_runtime))),
         })
     }