Add an example showcasing how to implement a custom network behavior and access streams

Cargo.toml

@@ -6,6 +6,7 @@ members = [
     "examples/chat",
     "examples/dcutr",
     "examples/distributed-key-value-store",
+    "examples/escape-hatch",
     "examples/file-sharing",
     "examples/identify",
     "examples/ipfs-kad",

examples/escape-hatch/Cargo.toml

@@ -0,0 +1,25 @@
+[package]
+name = "escape-hatch"
+version = "0.1.0"
+edition = "2021"
+publish = false
+license = "MIT"
+
+[package.metadata.release]
+release = false
+
+[dependencies]
+async-std = { version = "1.12", features = ["attributes"] }
+async-trait = "0.1"
+futures = "0.3.29"
+futures-timer = "3.0.2"
+libp2p = { path = "../../libp2p", features = [ "async-std", "dns", "kad", "mdns", "noise", "macros", "tcp", "yamux"] }
+libp2p-core = { workspace = true }
+rand = "0.8"
+tokio = { version = "1.33", features = ["macros", "net", "rt", "signal"] }
+tracing = "0.1.37"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+void = "1.0"
+
+[lints]
+workspace = true

examples/escape-hatch/src/handler.rs

@@ -0,0 +1,68 @@
+use libp2p::swarm::{ConnectionHandler, SubstreamProtocol, ConnectionHandlerEvent};
+use libp2p::StreamProtocol;
+use libp2p_core::upgrade::ReadyUpgrade;
+use std::task::Context;
+use std::{task::Poll, time::Duration};
+use void::Void;
+
+use crate::{Config, Error, PROTOCOL_NAME};
+
+pub struct Connection {
+    config: Config,
+}
+
+impl Connection {
+    pub fn new(config: Config) -> Self {
+        Self { config }
+    }
+}
+
+impl ConnectionHandler for Connection {
+    type FromBehaviour = Void;
+
+    type ToBehaviour = Result<Duration, Error>;
+
+    type InboundProtocol = ReadyUpgrade<StreamProtocol>;
+
+    type OutboundProtocol = ReadyUpgrade<StreamProtocol>;
+
+    type InboundOpenInfo = ();
+
+    type OutboundOpenInfo = ();
+
+    fn listen_protocol(
+        &self,
+    ) -> libp2p::swarm::SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
+        SubstreamProtocol::new(ReadyUpgrade::new(PROTOCOL_NAME), ())
+    }
+
+    fn on_behaviour_event(&mut self, _: Void) {}
+
+    fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<
+        libp2p::swarm::ConnectionHandlerEvent<
+            Self::OutboundProtocol,
+            Self::OutboundOpenInfo,
+            Self::ToBehaviour,
+        >,
+    > {
+        let protocol = self.listen_protocol();
+        return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest {
+            protocol,
+        });
+    }
+
+    fn on_connection_event(
+        &mut self,
+        event: libp2p::swarm::handler::ConnectionEvent<
+            Self::InboundProtocol,
+            Self::OutboundProtocol,
+            Self::InboundOpenInfo,
+            Self::OutboundOpenInfo,
+        >,
+    ) {
+        todo!()
+    }
+}

examples/escape-hatch/src/lib.rs

@@ -0,0 +1,173 @@
+use futures::channel::oneshot;
+use handler::Connection;
+use libp2p::swarm::{
+    ConnectionDenied, ConnectionId, FromSwarm, SubstreamProtocol, THandler, THandlerOutEvent,
+    ToSwarm,
+};
+use libp2p::{swarm::NetworkBehaviour, tcp::tokio::TcpStream, PeerId, Stream, StreamProtocol};
+use libp2p_core::upgrade::ReadyUpgrade;
+use libp2p_core::{Endpoint, Multiaddr};
+use std::task::Context;
+use std::{collections::VecDeque, io, task::Poll, time::Duration};
+use std::sync::mpsc;
+
+mod handler;
+
+pub const PROTOCOL_NAME: StreamProtocol = StreamProtocol::new("/ipfs/ping/1.0.0");
+
+#[derive(Debug)]
+pub enum Error {
+    NotConnected(PeerId),
+    UnsupportedProtocol,
+    Io(io::Error),
+}
+
+/// Event generated by the `Ping` network behaviour.
+#[derive(Debug)]
+pub struct Event {
+    /// The peer ID of the remote.
+    pub peer: PeerId,
+    /// The connection the ping was executed on.
+    pub connection: ConnectionId,
+    /// The result of an inbound or outbound ping.
+    pub result: Result<Duration, Error>,
+}
+
+/// A behaviour that manages requests to open new streams which are directly handed to the user.
+pub struct Behaviour {
+    /// Config
+    config: Config,
+    /// Queue of events to yield to the swarm.
+    events: VecDeque<Event>,
+    /// Protocol
+    protocol: StreamProtocol,
+}
+
+#[derive(Debug, Clone)]
+pub struct Config {
+    /// The timeout of an outbound ping.
+    timeout: Duration,
+    /// The duration between outbound pings.
+    interval: Duration,
+}
+
+struct StreamMessage {
+    peer_id: PeerId, // Assuming PeerId is a type representing a peer
+    response_channel: oneshot::Sender<Result<Stream, Error>>, // For sending back the result
+}
+
+pub struct IncomingStreams {
+    /// Queue of events to yield to the swarm.
+    events: VecDeque<Event>,
+}
+
+/// A control acts as a "remote" that allows users to request streams without interacting with the `Swarm` directly.
+
+pub struct Control {
+    sender: mpsc::Sender<StreamMessage>,
+    receiver: mpsc::Receiver<StreamMessage>
+}
+
+impl IncomingStreams {
+    pub async fn next(&mut self) -> (PeerId, Stream) {
+        if let Some(e) = self.events.pop_back() {
+        } else {
+        }
+
+        todo!()
+    }
+
+    pub fn poll_next(&mut self, cx: &mut Context<'_>) -> Poll<(PeerId, Stream)> {
+        todo!()
+    }
+}
+
+impl Behaviour {
+    pub fn new(protocol: StreamProtocol) -> (Self, Control, IncomingStreams) {
+        let (sender, receiver) = mpsc::channel::<StreamMessage>();
+        let behaviour = Self {
+            config: Config {
+                timeout: Duration::from_secs(1),
+                interval: Duration::from_secs(1),
+            },
+            events: VecDeque::new(),
+            protocol,
+        };
+
+        (
+            behaviour,
+            Control {
+                sender,
+                receiver
+            },
+            IncomingStreams {
+                events: VecDeque::new(),
+            },
+        )
+    }
+}
+
+impl Control {
+    pub async fn open_stream(&self, peer_id: PeerId) -> Result<Stream, Error> {
+        let (response_tx, response_rx) = oneshot::channel();
+        let message = StreamMessage {
+            peer_id,
+            response_channel: response_tx,
+        };
+        // Send the message to NetworkBehaviour
+        self.sender.send(message).expect("Failed to send message");
+        response_rx.await.expect("Failed to receive response")
+    }
+}
+
+impl NetworkBehaviour for Behaviour {
+    type ConnectionHandler = Connection;
+
+    type ToSwarm = Event;
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Connection::new(self.config.clone()))
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _: ConnectionId,
+        _: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Connection::new(self.config.clone()))
+    }
+
+    fn on_swarm_event(&mut self, _event: FromSwarm) {}
+
+    fn on_connection_handler_event(
+        &mut self,
+        peer: PeerId,
+        connection: ConnectionId,
+        result: THandlerOutEvent<Self>,
+    ) {
+        self.events.push_front(Event {
+            peer,
+            connection,
+            result,
+        })
+    }
+
+    fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<libp2p::swarm::ToSwarm<Self::ToSwarm, libp2p::swarm::THandlerInEvent<Self>>> {
+        if let Some(e) = self.events.pop_back() {
+            Poll::Ready(ToSwarm::GenerateEvent(e))
+        } else {
+            Poll::Pending
+        }
+    }
+}

examples/escape-hatch/src/main.rs

@@ -0,0 +1,15 @@
+use escape_hatch::Behaviour;
+use futures::prelude::*;
+use futures::StreamExt;
+use libp2p::StreamProtocol;
+use libp2p_core::{
+    multiaddr::multiaddr,
+    transport::{memory::MemoryTransport, ListenerId, Transport},
+};
+use rand::{thread_rng, Rng};
+
+#[tokio::main]
+async fn main() {
+    let stream_protocol = StreamProtocol::new("/escape-hatch/");
+    let (behaviour, control, incoming_stream) = Behaviour::new(stream_protocol);
+}

examples/metrics/Cargo.toml

@@ -20,6 +20,7 @@ tokio = { version = "1", features = ["full"] }
 tracing = "0.1.37"
 tracing-opentelemetry = "0.21.0"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-appender = "0.2"
 
 [lints]
 workspace = true