feat: transition from SubstreamSet to FuturesStream

[iTranscend]

in partial resolution of #342

This PR replaces the use of the SubstreamSet hashmap in substream's test with the FuturesStream type.

[lexnv]

Thanks @iTranscend for contributing 🙏

This looks good to me, it is nice that we have tests for our FuturesStream implementation

[dmitry-markin]

These tests were testing the SubstreamSet, implemented in this file. As it was replaced with FuturesStream, the equivalent tests should go to utils/futures_stream.rs.

Also, the SubstreamSet should be completely removed.

[iTranscend]

regarding the complete removal of SubstreamSet

Removing it will render the CustomProtocol test defunct as FuturesStream is not a good replacement for SubstreamSet in that case.

In the implementation of UserProtocol for CustomProtocol, when a command is received in the event loop, the PeerId is used as a key to find the corresponding Substream of the peer that the received command is directed at.

litep2p/tests/substream.rs

Lines 152 to 178 in 3f13b10

	command = self.rx.recv() => match command.unwrap() {
	Command::SendPayloadFramed(peer, payload, tx) => {
	match self.substreams.remove(&peer) {
	None => {
	tx.send(Err(Error::PeerDoesntExist(peer))).unwrap();
	}
	Some(mut substream) => {
	let payload = Bytes::from(payload);
	let res = substream.send_framed(payload).await.map_err(Into::into);
	tx.send(res).unwrap();
	let _ = substream.close().await;
	}
	}
	}
	Command::SendPayloadSink(peer, payload, tx) => {
	match self.substreams.remove(&peer) {
	None => {
	tx.send(Err(Error::PeerDoesntExist(peer))).unwrap();
	}
	Some(mut substream) => {
	let payload = Bytes::from(payload);
	let res = substream.send(payload).await.map_err(Into::into);
	tx.send(res).unwrap();
	let _ = substream.close().await;
	}
	}
	}

This behavior can't be replicated with FuturesStream because there is no way to know what future (containing the substream for the peer in focus) will be returned. We can't also iterate through the FuturesStream because that would resolve futures of substreams that we are not interested in.

My intuition on this may be wrong and I may be looking at this from the wrong angle so please lmk what you think.

cc @lexnv @dmitry-markin

[dmitry-markin]

@iTranscend Thanks for the explanation!

@lexnv What was the motivation behind the removal of SubstreamSet in the first place?

If it's used in the other parts of the codebase, we should either rewrite these parts without its use, or keep it with all the tests. May be finding a better place for it.

[lexnv]

@lexnv What was the motivation behind the removal of SubstreamSet in the first place?

The core logic of litep2p has moved away from SubstreamSet toward FuturesStream.
However, we do have a testing component CustomProtocol which still uses that.

Offhand, I believe we can adjust the logic to achieve a similar output:

We'd need to keep polling the futuresStream as before:

event = self.substreams.next()

There needs to be a mapping between the peer ID and the equivalent tokio channel for forwarding commands:

struct CustomProtocol {
mappings: HashMap<PeerId, Sender<Command>>,

When we populate the FuturesStream, we'd need to create a new channel:

let (tx, rx) = channel(1024);
self.mappings.insert(peer_id, tx);

let substream_work_futures = async move { ... };

self.substreams.push(async move {
loop {
   tokio::select {
       // Handle command
       command = rx => ... { }
       
       // Handle substream messages 
       message = substream.next() => {
       
          return Ok((peer, message, rx))
}

If this turns to be to complicated, we could also move SubstreamSet to the private testing module 🙏