Merge pull request #405 from nervosnetwork/support-session-init-from-user

feat: support session init from user

Author: driftluo

tentacle/src/service.rs

@@ -48,7 +48,7 @@ pub use crate::service::{
         HandshakeType, ProtocolHandle, ProtocolMeta, TargetProtocol, TargetSession, TcpSocket,
     },
     control::{ServiceAsyncControl, ServiceControl},
-    event::{ServiceError, ServiceEvent},
+    event::{RawSessionInfo, ServiceError, ServiceEvent},
     helper::SessionType,
 };
 use bytes::Bytes;
@@ -1150,6 +1150,23 @@ where
                     }
                 }
             }
+            ServiceTask::RawSession {
+                remote_address,
+                raw_session,
+                session_info,
+            } => {
+                let (ty, listen_addr) = match session_info {
+                    RawSessionInfo::Inbound { listen_addr } => {
+                        (SessionType::Inbound, Some(listen_addr))
+                    }
+                    RawSessionInfo::Outbound { target } => {
+                        self.dial_protocols.insert(remote_address.clone(), target);
+                        self.state.increase();
+                        (SessionType::Outbound, None)
+                    }
+                };
+                self.handshake(raw_session, ty, remote_address, listen_addr);
+            }
             ServiceTask::Disconnect { session_id } => {
                 self.session_close(session_id, Source::External).await
             }

tentacle/src/service/control.rs

@@ -1,13 +1,17 @@
 use futures::prelude::*;
 
+use std::fmt::Debug;
 use std::sync::{atomic::Ordering, Arc};
 use std::time::Duration;
 
 use crate::{
     channel::mpsc,
     error::SendErrorKind,
     multiaddr::Multiaddr,
-    service::{event::ServiceTask, TargetProtocol, TargetSession},
+    service::{
+        event::{RawSessionInfo, ServiceTask},
+        TargetProtocol, TargetSession,
+    },
     ProtocolId, SessionId,
 };
 use bytes::Bytes;
@@ -72,6 +76,25 @@ impl ServiceControl {
         self.quick_send(ServiceTask::Dial { address, target })
     }
 
+    /// Receive an established connection session
+    /// and build the tentacle protocol on top of it.
+    #[inline]
+    pub fn raw_session<T>(
+        &self,
+        raw_session: T,
+        remote_address: Multiaddr,
+        info: RawSessionInfo,
+    ) -> Result
+    where
+        T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
+    {
+        self.quick_send(ServiceTask::RawSession {
+            raw_session: Box::new(raw_session),
+            remote_address,
+            session_info: info,
+        })
+    }
+
     /// Disconnect a connection
     #[inline]
     pub fn disconnect(&self, session_id: SessionId) -> Result {
@@ -255,6 +278,18 @@ impl From<ServiceAsyncControl> for ServiceControl {
     }
 }
 
+impl Debug for ServiceControl {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "ServiceControl")
+    }
+}
+
+impl Debug for ServiceAsyncControl {
+    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
+        write!(f, "ServiceAsyncControl")
+    }
+}
+
 /// Service control, used to send commands externally at runtime, All interfaces are async methods
 #[derive(Clone)]
 pub struct ServiceAsyncControl {
@@ -301,6 +336,26 @@ impl ServiceAsyncControl {
         self.quick_send(ServiceTask::Dial { address, target }).await
     }
 
+    /// Receive an established connection session
+    /// and build the tentacle protocol on top of it.
+    #[inline]
+    pub async fn raw_session<T>(
+        &self,
+        raw_session: T,
+        remote_address: Multiaddr,
+        info: RawSessionInfo,
+    ) -> Result
+    where
+        T: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
+    {
+        self.quick_send(ServiceTask::RawSession {
+            raw_session: Box::new(raw_session),
+            remote_address,
+            session_info: info,
+        })
+        .await
+    }
+
     /// Disconnect a connection
     #[inline]
     pub async fn disconnect(&self, session_id: SessionId) -> Result {

tentacle/src/service/event.rs

@@ -228,10 +228,46 @@ pub(crate) enum ServiceTask {
         /// Listen address
         address: Multiaddr,
     },
+    /// Receive an established connection session
+    /// and build the tentacle protocol on top of it.
+    RawSession {
+        /// Remote address
+        remote_address: Multiaddr,
+        /// Raw session
+        raw_session: Box<dyn crate::session::AsyncRw + Send + Unpin + 'static>,
+        /// Session info
+        session_info: RawSessionInfo,
+    },
     /// Shutdown service
     Shutdown(bool),
 }
 
+/// Raw session info
+pub enum RawSessionInfo {
+    /// Inbound session
+    Inbound {
+        /// Session provenance
+        listen_addr: Multiaddr,
+    },
+    /// Outbound session
+    Outbound {
+        /// Dial protocols
+        target: TargetProtocol,
+    },
+}
+
+impl RawSessionInfo {
+    /// Inbound session info
+    pub fn inbound(listen_addr: Multiaddr) -> Self {
+        RawSessionInfo::Inbound { listen_addr }
+    }
+
+    /// Outbound session info
+    pub fn outbound(target: TargetProtocol) -> Self {
+        RawSessionInfo::Outbound { target }
+    }
+}
+
 impl fmt::Debug for ServiceTask {
     fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
         use self::ServiceTask::*;
@@ -269,6 +305,7 @@ impl fmt::Debug for ServiceTask {
             Disconnect { session_id } => write!(f, "Disconnect session [{}]", session_id),
             Dial { address, .. } => write!(f, "Dial address: {}", address),
             Listen { address } => write!(f, "Listen address: {}", address),
+            RawSession { remote_address, .. } => write!(f, "Raw session from: {}", remote_address),
             ProtocolOpen { session_id, .. } => write!(f, "Open session [{}] proto", session_id),
             ProtocolClose {
                 session_id,