Direct messages and arbitrary data transfer between peers.

[sandeepmukh]

Hi! I'm new to libp2p (and rust), so I'm working through the examples. Right now I'm working through the chat-tokio.rs example and I think I understand how it works pretty well. There are a couple of things that I'm confused about doing:

1. If I wanted to send arbitrary data, say a JSON message to the channel, would an approach like this work? Obviously, it's a bit silly because (I think a least lol) it would continually send the same file, but imagine that the JSON are being served up from some background process.

    loop {
        tokio::select! {

            line = json::stringify(jsonFile)? => { // This syntax is probably wrong, but that's not the point
                swarm.behaviour_mut().floodsub.publish(floodsub_topic.clone(), line.as_bytes());
            }
            event = swarm.select_next_some() => {
                if let SwarmEvent::NewListenAddr { address, .. } = event {
                    local_address = address;
                    println!("Listening on {:?}", local_address);
                }
            }
        }
    }

2. How would I directly message a peer node given its peerID? Let imagine that I'm trying to send some information to a peer that I don't want every node to be able to see--how would I achieve this? I tried looking into using the Connection trait, but I couldn't figure out how to implement it. Would this require some sort of new network behavior?

Any help would be greatly appreciated!

[thomaseizinger]

1. How would I directly message a peer node given its peerID? Let imagine that I'm trying to send some information to a peer that I don't want every node to be able to see--how would I achieve this? I tried looking into using the `Connection` trait, but I couldn't figure out how to implement it. Would this require some sort of new network behavior?

Yes. If you just want to send a single message with an optional response then you can use libp2p-request-response.

Anything more elaborate will require implementing your own ProtocolsHandler and NetworkBehaviour.

If I wanted to send arbitrary data, say a JSON message to the channel, would an approach like this work? Obviously, it's a bit silly because (I think a least lol) it would continually send the same file, but imagine that the JSON are being served up from some background process.

That would publish the JSON under this topic yes. All peers subscribed to this topic and currently online would receive the message.

If I wanted to send arbitrary data, say a JSON message to the channel, would an approach like this work?

Rust-libp2p and libp2p in general provides a lot of things out-of-the-box like authentication, multiplexing etc.
Hence, thinking in terms of "arbitrary data" is in my experience not particularly helpful. Instead, you want to think in "protocols" that achieve a certain goal and then implement on NetworkBehaviour per protocol.

I can recommend reading through some of the libp2p specs to get an idea: https://github.com/libp2p/specs

For example, there is the /identify protocol which - on a technical level - is just request-response but really, what it achieves functionally is to gather information about all connected peers.
Same thing for ping; again it is request-response but the functionality it achieves is detecting dead connections and measuring latency.
Another example would be rendezvous: Also request-response but its purpose is to register with a rendezvous point that you are providing some sort of service and other nodes can discover you.

[sandeepmukh]

Thanks for your quick response! I'm trying to implement a swarm with both the floodsub and request_response protocols, but I'm having a bit of trouble doing it. From what I understand so far, RequestResponse requires a RequestResponseCodec which is a trait. I'm trying to implement a codec that sends stringified JSON (as u8 vectors) between nodes.

From what I understand, I'll have to pass in some sort of io communication like a tokio TCPSocket (is this the case?). From there I'll use serde_json to read the json from the stream. Here's what I have so far:

I've taken inspiration from the test cases in the repo.

#[derive(Debug, Clone)]
struct MsgProtocol();
#[derive(Clone)]
struct MsgCodec();
#[derive(Debug, Clone, PartialEq, Eq)]
struct In(serde_json::Value);
#[derive(Debug, Clone, PartialEq, Eq)]
struct Out(Vec<u8>);

impl ProtocolName for MsgProtocol {
    fn protocol_name(&self) -> &[u8] {
        "/msg/1".as_bytes()
    }
}

#[async_trait]
impl RequestResponseCodec for MsgCodec {
    type Protocol = MsgProtocol;
    type Request = In;
    type Response = Out;

    async fn read_request<T>(&mut self, _: &MsgProtocol, io: &mut T)
        -> io::Result<Self::Request>
    where
        T: AsyncRead + Unpin + Send
    {
        let json = serde_json::from_reader(//what would go here? I think it would be a TcpStream, but I only have an io socket);

        if json == serde_json::json!(null) {
            return Err(io::ErrorKind::UnexpectedEof.into())
        }

        Ok(In(json))
    }
        async fn write_request<T>(
        &mut self,
        _: &MsgProtocol,
        io: &mut T,
        In(data): In,
    ) -> io::Result<()>
    where
        T: AsyncWrite + Unpin + Send,
    {
        write_length_prefixed(io, data).await?;
        io.close().await?;

        Ok(())
    }
}

[mxinden]

From what I understand, I'll have to pass in some sort of io communication like a tokio TCPSocket (is this the case?).

No, that is taken care of by libp2p.

The simple Ping example should help getting started.

rust-libp2p/protocols/request-response/tests/ping.rs

Lines 73 to 155 in e437c00

	/// Exercises a simple ping protocol.
	#[test]
	fn ping_protocol() {
	let ping = Ping("ping".to_string().into_bytes());
	let pong = Pong("pong".to_string().into_bytes());
	let protocols = iter::once((PingProtocol(), ProtocolSupport::Full));
	let cfg = RequestResponseConfig::default();
	let (peer1_id, trans) = mk_transport();
	let ping_proto1 = RequestResponse::new(PingCodec(), protocols.clone(), cfg.clone());
	let mut swarm1 = Swarm::new(trans, ping_proto1, peer1_id);
	let (peer2_id, trans) = mk_transport();
	let ping_proto2 = RequestResponse::new(PingCodec(), protocols, cfg);
	let mut swarm2 = Swarm::new(trans, ping_proto2, peer2_id);
	let (mut tx, mut rx) = mpsc::channel::<Multiaddr>(1);
	let addr = "/ip4/127.0.0.1/tcp/0".parse().unwrap();
	swarm1.listen_on(addr).unwrap();
	let expected_ping = ping.clone();
	let expected_pong = pong.clone();
	let peer1 = async move {
	loop {
	match swarm1.select_next_some().await {
	SwarmEvent::NewListenAddr { address, .. }=> tx.send(address).await.unwrap(),
	SwarmEvent::Behaviour(RequestResponseEvent::Message {
	peer,
	message: RequestResponseMessage::Request { request, channel, .. }
	}) => {
	assert_eq!(&request, &expected_ping);
	assert_eq!(&peer, &peer2_id);
	swarm1.behaviour_mut().send_response(channel, pong.clone()).unwrap();
	},
	SwarmEvent::Behaviour(RequestResponseEvent::ResponseSent {
	peer, ..
	}) => {
	assert_eq!(&peer, &peer2_id);
	}
	SwarmEvent::Behaviour(e) => panic!("Peer1: Unexpected event: {:?}", e),
	_ => {}
	}
	}
	};
	let num_pings: u8 = rand::thread_rng().gen_range(1, 100);
	let peer2 = async move {
	let mut count = 0;
	let addr = rx.next().await.unwrap();
	swarm2.behaviour_mut().add_address(&peer1_id, addr.clone());
	let mut req_id = swarm2.behaviour_mut().send_request(&peer1_id, ping.clone());
	assert!(swarm2.behaviour().is_pending_outbound(&peer1_id, &req_id));
	loop {
	match swarm2.select_next_some().await {
	SwarmEvent::Behaviour(RequestResponseEvent::Message {
	peer,
	message: RequestResponseMessage::Response { request_id, response }
	}) => {
	count += 1;
	assert_eq!(&response, &expected_pong);
	assert_eq!(&peer, &peer1_id);
	assert_eq!(req_id, request_id);
	if count >= num_pings {
	return
	} else {
	req_id = swarm2.behaviour_mut().send_request(&peer1_id, ping.clone());
	}
	}
	SwarmEvent::Behaviour(e) =>panic!("Peer2: Unexpected event: {:?}", e),
	_ => {}
	}
	}
	};
	async_std::task::spawn(Box::pin(peer1));
	let () = async_std::task::block_on(peer2);
	}

[sandeepmukh]

Hm I'm having difficulty replicating the ping example. the io.close().await?; line from the write_* methods seems to not want to compile. Have you run into an issue like this?

Edit: just forgot some imports! Nevermind

[elenaf9]

I'm trying to implement a codec that sends stringified JSON (as u8 vectors) between nodes.

In case that it helps, we did something similar in our project: https://github.com/iotaledger/stronghold.rs/blob/322a6b6960ba4d8b47db459b2f66745d8464a3e7/communication/src/behaviour/protocol.rs#L50

• The basic requirement for the request/ response messages is that they implement Serialize and Deserialize (and also 'static I think)
  In our case we enforced it with trait bounds, since we use generic Request/ Response messages.

• For writing requests/ responses to the substream, the message is serialized via serde_json::to_vec and written to the substream with write_length_prefixed.
  (Note: my linked example uses an older version of libp2p with write_one, this is outdated, use write_length_prefixed instead).

• For reading requests/responses from the substream, it is attempted to parse the message with serde_json::from_slice.
  (Note again: we use the outdated read_one, but read_length_prefixed should be used instead).

• Errors in serializing / deserializing are mapped into a IoError::new(IoErrorKind::InvalidData, e).

[elenaf9]

Ah nevermind, I just noticed that I misunderstood your question. You are not trying to serialize/ deserialize a request/response message, but instead read from a json file and send its content. Then of course the file that I linked won't be much of a help.

[elv-sandeep]

No, it was certainly useful to see another example! Thanks :)

[sandeepmukh]

I've been able to replicate the request/response protocol for my data, but I now need to get the message out the inject_event function to be fed given to a process that is external to the library, like a crossbeam_channel that's feeding data to something. Is this possible? If so how?

[thomaseizinger]

My personal preference is to use #[behaviour(event_process = false)] on the NetworkBehaviour. Then you get all your events when polling the swarm through swarm.next() as part of SwarmEvent::Behaviour.

[sandeepmukh]

For some reason this isnt getting me floodsub and request_response events—is there another way?

Would it be possible to add a queue to my network that I could push events I care about to and then poll it in the event loop?

[thomaseizinger]

Does this help?

#2171

[sandeepmukh]

It does thank you! Your help has been absolutely invaluable :).