Issue with Opening Stream after Dialing Receiver

[vinay10949]

I have managed to set up a relay, client, and receiver using the libp2p framework. I am able to successfully dial the receiver, but I'm encountering an issue when trying to open a stream to communicate data.

Here's a snippet of my code related to the issue:

use clap::Parser;
use futures::{executor::block_on, future::FutureExt, stream::StreamExt, AsyncReadExt, AsyncWriteExt};
use libp2p::{
    core::multiaddr::{Multiaddr, Protocol},
    dcutr,
    identify,
    identity,
    ping,
    relay,
    noise,
    swarm::{NetworkBehaviour, SwarmEvent},
    tcp, yamux, PeerId,
};
use std::{io::{self, Write}};
use std::str::FromStr;
use std::{error::Error, time::Duration};
use tracing_subscriber::EnvFilter;

#[derive(Debug, Parser)]
#[clap(name = "libp2p DCUtR client")]
struct Opts {
    /// The mode (client-listen, client-dial).
    #[clap(long)]
    mode: Mode,

    /// Fixed value to generate deterministic peer id.
    #[clap(long)]
    secret_key_seed: u8,

    /// The listening address
    #[clap(long)]
    relay_address: Multiaddr,

    /// Peer ID of the remote peer to hole punch to.
    #[clap(long)]
    remote_peer_id: Option<PeerId>,
}

#[derive(Clone, Debug, PartialEq, Parser)]
enum Mode {
    Dial,
    Listen,
}

impl FromStr for Mode {
    type Err = String;
    fn from_str(mode: &str) -> Result<Self, Self::Err> {
        match mode {
            "dial" => Ok(Mode::Dial),
            "listen" => Ok(Mode::Listen),
            _ => Err("Expected either 'dial' or 'listen'".to_string()),
        }
    }
}


use libp2p_swarm::StreamProtocol;
const TRANSFER: StreamProtocol = StreamProtocol::new("/transfer");

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .try_init();

    let opts = Opts::parse();

    #[derive(NetworkBehaviour)]
    struct Behaviour {
        relay_client: relay::client::Behaviour,
        ping: ping::Behaviour,
        identify: identify::Behaviour,
        dcutr: dcutr::Behaviour,
    }

    let mut swarm = libp2p::SwarmBuilder::with_existing_identity(generate_ed25519(opts.secret_key_seed))
        .with_tokio()
        .with_tcp(tcp::Config::default().port_reuse(true).nodelay(true),noise::Config::new, yamux::Config::default)?
        .with_quic()
        .with_dns()?
        .with_relay_client(noise::Config::new, yamux::Config::default)?
        .with_behaviour(|keypair, relay_behaviour| {
            println!("Peer key: {:?}", keypair.public().to_peer_id());
            Behaviour {
                relay_client: relay_behaviour,
                ping: ping::Behaviour::new(ping::Config::new()),
                identify: identify::Behaviour::new(identify::Config::new(
                    "/TODO/0.0.1".to_string(),
                    keypair.public(),
                )),
                dcutr: dcutr::Behaviour::new(keypair.public().to_peer_id()),
            }
        })?
        .with_swarm_config(|c| c.with_idle_connection_timeout(Duration::from_secs(60)))
        .build();

    let mut incoming_streams = libp2p_stream::Behaviour::new().new_control().accept(TRANSFER).unwrap();

    swarm
        .listen_on("/ip4/0.0.0.0/udp/0/quic-v1".parse().unwrap())
        .unwrap();
    swarm
        .listen_on("/ip4/0.0.0.0/tcp/0".parse().unwrap())
        .unwrap();

    block_on(async {
        let mut delay = futures_timer::Delay::new(std::time::Duration::from_secs(1)).fuse();
        loop {
            futures::select! {
                event = swarm.next() => {
                    match event.unwrap() {
                        SwarmEvent::NewListenAddr { address, .. } => {
                            tracing::info!(%address, "Listening on address");
                        }
                        event => panic!("{event:?}"),
                    }
                }
                _ = delay => {
                    break;
                }
            }
        }
    });

    swarm.dial(opts.relay_address.clone()).unwrap();
    block_on(async {
        let mut learned_observed_addr = false;
        let mut told_relay_observed_addr = false;

        loop {
            match swarm.next().await.unwrap() {
                SwarmEvent::NewListenAddr { .. } => {}
                SwarmEvent::Dialing { .. } => {}
                SwarmEvent::ConnectionEstablished { .. } => {}
                SwarmEvent::Behaviour(BehaviourEvent::Ping(_)) => {}
                SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Sent {
                    ..
                })) => {
                    tracing::info!("Told relay its public address");
                    told_relay_observed_addr = true;
                }
                SwarmEvent::Behaviour(BehaviourEvent::Identify(identify::Event::Received {
                    info: identify::Info { observed_addr, .. },
                    ..
                })) => {
                    tracing::info!(address=%observed_addr, "Relay told us our observed address");
                    learned_observed_addr = true;
                }
                event => panic!("{event:?}"),
            }

            if learned_observed_addr && told_relay_observed_addr {
                break;
            }
        }
    });

    match opts.mode {
        Mode::Dial => {
            let remote_peer_id = opts.remote_peer_id.unwrap();
           let mult_addr=opts.relay_address.with(Protocol::P2pCircuit).with(Protocol::P2p(remote_peer_id));
           let status= swarm.dial(mult_addr.clone()).unwrap();
        
            let mut stream = libp2p_stream::Behaviour::new().new_control().open_stream(remote_peer_id, TRANSFER).await.unwrap();
            println!("Stream ");
            let stdin_task = tokio::spawn(async move {
                loop {
                    let mut input = String::new();
                    print!("Enter message: ");
                    io::stdout().flush().unwrap();
                    io::stdin().read_line(&mut input).unwrap();
                    let input = input.trim();
                    if input == "/quit" {
                        break;
                    }
                    stream.write_all(input.as_bytes()).await.unwrap();
                }
            });
            
        }
        Mode::Listen => {
            swarm.listen_on(opts.relay_address.with(Protocol::P2pCircuit)).unwrap();
            println!("LISTENING");
            tokio::spawn(async move {
                while let Some((_, mut stream)) = incoming_streams.next().await {
                    let mut buf = [0u8; 1000];
                    loop {
                        let read = stream.read(&mut buf).await.unwrap();
                        if read == 0 {
                            break;
                        }
                        if let Ok(text) = std::str::from_utf8(&buf[..read]) {
                            println!("Received: {}", text);
                        }
                    }
                }
            });
        }
    
    }

    block_on(async {
        loop {
            match swarm.next().await.unwrap() {
                SwarmEvent::NewListenAddr { address, .. } => {

                    let listen_address = address.with_p2p(*swarm.local_peer_id()).unwrap();
                    tracing::info!(%listen_address);
                    tracing::info!(%listen_address, "Listening on address");
                }
                SwarmEvent::Behaviour(BehaviourEvent::RelayClient(
                    relay::client::Event::ReservationReqAccepted { .. },
                )) => {
                    assert!(opts.mode == Mode::Listen);
                    tracing::info!("Relay accepted our reservation request");
                }
                SwarmEvent::Behaviour(BehaviourEvent::RelayClient(event)) => {
                    tracing::info!(?event)
                }
                SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => {
                    tracing::info!(?event)
                }
                SwarmEvent::Behaviour(BehaviourEvent::Identify(event)) => {
                    tracing::info!(?event)
                }
                SwarmEvent::Behaviour(BehaviourEvent::Ping(_)) => {}
                SwarmEvent::ConnectionEstablished {
                    peer_id, endpoint, ..
                } => {
                    tracing::info!(peer=%peer_id, ?endpoint, "Established new connection");
                }
                SwarmEvent::OutgoingConnectionError { peer_id, error, .. } => {
                    tracing::info!(peer=?peer_id, "Outgoing connection failed: {error}");
                }
                _ => {}
            }
        }
    })
}

fn generate_ed25519(secret_key_seed: u8) -> identity::Keypair {
    let mut bytes = [0u8; 32];
    bytes[0] = secret_key_seed;

    identity::Keypair::ed25519_from_bytes(bytes).expect("only errors on wrong length")
}
        

Problem Statement:

After dialing the receiver using swarm.dial, the code hangs when trying to open a stream using open_stream. I have ensured that the remote_peer_id and TRANSFER protocol are correctly set up.

Expected Behavior:

I expect the code to successfully open a stream to the receiver after dialing, allowing data communication between the client and receiver.

Steps Taken:

1. Checked and verified that remote_peer_id and TRANSFER protocol are correctly set up.
2. Ensured that the necessary behaviours and configurations for opening streams are in place.
3. Verified that the dialing process to the receiver is successful.

Questions:

1. What could be causing the code to hang when opening the stream after dialing?
2. Are there any specific configurations or additional steps required to successfully open a stream post-dialing?

Any insights or suggestions would be greatly appreciated. Thank you!

[dariusc93]

The problem here is that youre using libp2p-stream outside of swarm. You would need to add it apart of your Behaviour first.

[vinay10949]

its says this I am make it as part of swarm the trait bound libp2p_stream::Behaviour: libp2p::libp2p_swarm::NetworkBehaviour is not satisfied

[JKincorperated]

Did you find a solution to this? I am having the same issue.

[dariusc93]

@JKincorperated could you post your code?

[JKincorperated]

I manage to fix it by copying the libp2p-stream crates' code locally and just fixing the issues the rust compiler raised.