Merge pull request #388 from nervosnetwork/fix-dial-inner-insert-order

fix: bug on unsupported protocol

tentacle/src/service.rs

@@ -357,8 +357,8 @@ where
     /// Use by inner
     #[inline(always)]
     fn dial_inner(&mut self, address: Multiaddr, target: TargetProtocol) -> Result<()> {
-        self.dial_protocols.insert(address.clone(), target);
         let dial_future = self.multi_transport.clone().dial(address.clone())?;
+        self.dial_protocols.insert(address.clone(), target);
 
         let handshake_type = self.handshake_type.clone();
         let timeout = self.config.timeout;

tentacle/tests/test_dial_unsupported_order.rs

@@ -0,0 +1,141 @@
+use std::{
+    sync::{
+        atomic::{AtomicU8, Ordering},
+        mpsc::{self, Sender},
+        Arc,
+    },
+    thread,
+    time::Duration,
+};
+
+use futures::channel;
+use tentacle::{
+    async_trait,
+    builder::{MetaBuilder, ServiceBuilder},
+    context::{ProtocolContext, ServiceContext},
+    multiaddr::{Multiaddr, Protocol},
+    secio::SecioKeyPair,
+    service::{ProtocolHandle, ProtocolMeta, Service, ServiceError, TargetProtocol},
+    traits::{ServiceHandle, ServiceProtocol},
+    ProtocolId,
+};
+
+pub fn create<F>(secio: bool, meta: ProtocolMeta, shandle: F) -> Service<F, SecioKeyPair>
+where
+    F: ServiceHandle + Unpin + 'static,
+{
+    let builder = ServiceBuilder::default()
+        .insert_protocol(meta)
+        .forever(true);
+
+    if secio {
+        builder
+            .handshake_type(SecioKeyPair::secp256k1_generated().into())
+            .build(shandle)
+    } else {
+        builder.build(shandle)
+    }
+}
+
+fn create_meta(id: impl Into<ProtocolId> + Copy + Send + 'static) -> ProtocolMeta {
+    MetaBuilder::new()
+        .id(id.into())
+        .service_handle(move || {
+            if id.into() == 0.into() {
+                ProtocolHandle::None
+            } else {
+                let handle = Box::new(PHandle);
+                ProtocolHandle::Callback(handle)
+            }
+        })
+        .build()
+}
+
+struct PHandle;
+
+#[async_trait]
+impl ServiceProtocol for PHandle {
+    async fn init(&mut self, _context: &mut ProtocolContext) {}
+}
+
+struct SHandle {
+    count: Arc<AtomicU8>,
+    sender: Sender<()>,
+}
+
+#[async_trait]
+impl ServiceHandle for SHandle {
+    async fn handle_error(&mut self, _control: &mut ServiceContext, event: ServiceError) {
+        if let ServiceError::DialerError { .. } = event {
+            let prv = self
+                .count
+                .fetch_add(1, std::sync::atomic::Ordering::Acquire);
+            if prv == 1 {
+                self.sender.send(()).unwrap();
+            }
+        }
+    }
+}
+
+fn test_dial_unsupport_protocol_order(secio: bool) {
+    let meta = create_meta(1);
+    let (addr_sender, addr_receiver) = channel::oneshot::channel::<Multiaddr>();
+    let count = Arc::new(AtomicU8::new(0));
+    let count_clone = count.clone();
+    let (tx, rx) = mpsc::channel();
+
+    thread::spawn(move || {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let mut service = create(secio, meta, ());
+        rt.block_on(async move {
+            let listen_addr = service
+                .listen("/ip4/127.0.0.1/tcp/0".parse().unwrap())
+                .await
+                .unwrap();
+            let _res = addr_sender.send(listen_addr);
+            service.run().await
+        });
+    });
+
+    let meta = create_meta(1);
+
+    thread::spawn(move || {
+        let rt = tokio::runtime::Runtime::new().unwrap();
+        let mut service = create(
+            secio,
+            meta,
+            SHandle {
+                count: count_clone,
+                sender: tx,
+            },
+        );
+        let control = service.control().clone();
+        rt.block_on(async move {
+            let mut listen_addr = addr_receiver.await.unwrap();
+            listen_addr.push(Protocol::Wss);
+            control
+                .dial(listen_addr.clone(), TargetProtocol::All)
+                .await
+                .unwrap();
+            control
+                .dial(listen_addr, TargetProtocol::All)
+                .await
+                .unwrap();
+            service.run().await
+        });
+    });
+
+    rx.recv_timeout(Duration::from_secs(10)).unwrap();
+
+    assert_eq!(count.load(Ordering::Acquire), 2);
+}
+
+#[test]
+fn test_dial_unsupport_protocol_order_with_secio() {
+    test_dial_unsupport_protocol_order(true)
+}
+
+#[test]
+fn test_dial_unsupport_protocol_order_with_no_secio() {
+    test_dial_unsupport_protocol_order(false)
+}