Merge branch 'main' into interop/py-rust

Author: seetadev

examples/ping/ping.py

@@ -57,7 +57,7 @@ async def send_ping(stream: INetStream) -> None:
 async def run(port: int, destination: str) -> None:
     localhost_ip = "127.0.0.1"
     listen_addr = multiaddr.Multiaddr(f"/ip4/0.0.0.0/tcp/{port}")
-    host = new_host()
+    host = new_host(listen_addrs=[listen_addr])
 
     async with host.run(listen_addrs=[listen_addr]), trio.open_nursery() as nursery:
         if not destination:

libp2p/__init__.py

@@ -1,5 +1,6 @@
 from collections.abc import (
     Mapping,
+    Sequence,
 )
 from importlib.metadata import version as __version
 from typing import (
@@ -9,6 +10,8 @@
     cast,
 )
 
+import multiaddr
+
 from libp2p.abc import (
     IHost,
     IMuxedConn,
@@ -154,6 +157,7 @@ def new_swarm(
     sec_opt: Optional[TSecurityOptions] = None,
     peerstore_opt: Optional[IPeerStore] = None,
     muxer_preference: Optional[Literal["YAMUX", "MPLEX"]] = None,
+    listen_addrs: Optional[Sequence[multiaddr.Multiaddr]] = None,
 ) -> INetworkService:
     """
     Create a swarm instance based on the parameters.
@@ -163,6 +167,7 @@ def new_swarm(
     :param sec_opt: optional choice of security upgrade
     :param peerstore_opt: optional peerstore
     :param muxer_preference: optional explicit muxer preference
+    :param listen_addrs: optional list of multiaddrs to listen on
     :return: return a default swarm instance
 
     Note: Yamux (/yamux/1.0.0) is the preferred stream multiplexer
@@ -175,8 +180,16 @@ def new_swarm(
 
     id_opt = generate_peer_id_from(key_pair)
 
-    # TODO: Parse `listen_addrs` to determine transport
-    transport = TCP()
+    if listen_addrs is None:
+        transport = TCP()
+    else:
+        addr = listen_addrs[0]
+        if addr.__contains__("tcp"):
+            transport = TCP()
+        elif addr.__contains__("quic"):
+            raise ValueError("QUIC not yet supported")
+        else:
+            raise ValueError(f"Unknown transport in listen_addrs: {listen_addrs}")
 
     # Generate X25519 keypair for Noise
     noise_key_pair = create_new_x25519_key_pair()
@@ -229,6 +242,7 @@ def new_host(
     peerstore_opt: Optional[IPeerStore] = None,
     disc_opt: Optional[IPeerRouting] = None,
     muxer_preference: Optional[Literal["YAMUX", "MPLEX"]] = None,
+    listen_addrs: Sequence[multiaddr.Multiaddr] = None,
 ) -> IHost:
     """
     Create a new libp2p host based on the given parameters.
@@ -239,6 +253,7 @@ def new_host(
     :param peerstore_opt: optional peerstore
     :param disc_opt: optional discovery
     :param muxer_preference: optional explicit muxer preference
+    :param listen_addrs: optional list of multiaddrs to listen on
     :return: return a host instance
     """
     swarm = new_swarm(
@@ -247,6 +262,7 @@ def new_host(
         sec_opt=sec_opt,
         peerstore_opt=peerstore_opt,
         muxer_preference=muxer_preference,
+        listen_addrs=listen_addrs,
     )
 
     if disc_opt is not None:

libp2p/abc.py

@@ -8,10 +8,14 @@
     KeysView,
     Sequence,
 )
+from types import (
+    TracebackType,
+)
 from typing import (
     TYPE_CHECKING,
     Any,
     AsyncContextManager,
+    Optional,
 )
 
 from multiaddr import (
@@ -215,7 +219,7 @@ async def accept_stream(self) -> "IMuxedStream":
         """
 
 
-class IMuxedStream(ReadWriteCloser):
+class IMuxedStream(ReadWriteCloser, AsyncContextManager["IMuxedStream"]):
     """
     Interface for a multiplexed stream.
 
@@ -249,6 +253,20 @@ def set_deadline(self, ttl: int) -> bool:
             otherwise False.
         """
 
+    @abstractmethod
+    async def __aenter__(self) -> "IMuxedStream":
+        """Enter the async context manager."""
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> None:
+        """Exit the async context manager and close the stream."""
+        await self.close()
+
 
 # -------------------------- net_stream interface.py --------------------------
 

libp2p/host/basic_host.py

@@ -200,6 +200,29 @@ async def new_stream(
         net_stream.set_protocol(selected_protocol)
         return net_stream
 
+    async def send_command(self, peer_id: ID, command: str) -> list[str]:
+        """
+        Send a multistream-select command to the specified peer and return
+        the response.
+
+        :param peer_id: peer_id that host is connecting
+        :param command: supported multistream-select command (e.g., "ls)
+        :raise StreamFailure: If the stream cannot be opened or negotiation fails
+        :return: list of strings representing the response from peer.
+        """
+        new_stream = await self._network.new_stream(peer_id)
+
+        try:
+            response = await self.multiselect_client.query_multistream_command(
+                MultiselectCommunicator(new_stream), command
+            )
+        except MultiselectClientError as error:
+            logger.debug("fail to open a stream to peer %s, error=%s", peer_id, error)
+            await new_stream.reset()
+            raise StreamFailure(f"failed to open a stream to peer {peer_id}") from error
+
+        return response
+
     async def connect(self, peer_info: PeerInfo) -> None:
         """
         Ensure there is a connection between this host and the peer

libp2p/protocol_muxer/multiselect.py

@@ -60,8 +60,14 @@ async def negotiate(
                 raise MultiselectError() from error
 
             if command == "ls":
-                # TODO: handle ls command
-                pass
+                supported_protocols = list(self.handlers.keys())
+                response = "\n".join(supported_protocols) + "\n"
+
+                try:
+                    await communicator.write(response)
+                except MultiselectCommunicatorError as error:
+                    raise MultiselectError() from error
+
             else:
                 protocol = TProtocol(command)
                 if protocol in self.handlers:

libp2p/protocol_muxer/multiselect_client.py

@@ -81,6 +81,36 @@ async def select_one_of(
 
         raise MultiselectClientError("protocols not supported")
 
+    async def query_multistream_command(
+        self, communicator: IMultiselectCommunicator, command: str
+    ) -> list[str]:
+        """
+        Send a multistream-select command over the given communicator and return
+        parsed response.
+
+        :param communicator: communicator to use to communicate with counterparty
+        :param command: supported multistream-select command(e.g., ls)
+        :raise MultiselectClientError: If the communicator fails to process data.
+        :return: list of strings representing the response from peer.
+        """
+        await self.handshake(communicator)
+
+        if command == "ls":
+            try:
+                await communicator.write("ls")
+            except MultiselectCommunicatorError as error:
+                raise MultiselectClientError() from error
+        else:
+            raise ValueError("Command not supported")
+
+        try:
+            response = await communicator.read()
+            response_list = response.strip().splitlines()
+        except MultiselectCommunicatorError as error:
+            raise MultiselectClientError() from error
+
+        return response_list
+
     async def try_select(
         self, communicator: IMultiselectCommunicator, protocol: TProtocol
     ) -> TProtocol:

libp2p/pubsub/gossipsub.py

@@ -10,6 +10,7 @@
 )
 import logging
 import random
+import time
 from typing import (
     Any,
     DefaultDict,
@@ -80,8 +81,7 @@ class GossipSub(IPubsubRouter, Service):
     # The protocol peer supports
     peer_protocol: dict[ID, TProtocol]
 
-    # TODO: Add `time_since_last_publish`
-    #   Create topic --> time since last publish map.
+    time_since_last_publish: dict[str, int]
 
     mcache: MessageCache
 
@@ -138,6 +138,7 @@ def __init__(
             self.direct_peers[direct_peer.peer_id] = direct_peer
         self.direct_connect_interval = direct_connect_interval
         self.direct_connect_initial_delay = direct_connect_initial_delay
+        self.time_since_last_publish = {}
 
     async def run(self) -> None:
         if self.pubsub is None:
@@ -253,6 +254,8 @@ async def publish(self, msg_forwarder: ID, pubsub_msg: rpc_pb2.Message) -> None:
             except StreamClosed:
                 logger.debug("Fail to publish message to %s: stream closed", peer_id)
                 self.pubsub._handle_dead_peer(peer_id)
+        for topic in pubsub_msg.topicIDs:
+            self.time_since_last_publish[topic] = int(time.time())
 
     def _get_peers_to_send(
         self, topic_ids: Iterable[str], msg_forwarder: ID, origin: ID
@@ -342,6 +345,7 @@ async def join(self, topic: str) -> None:
             await self.emit_graft(topic, peer)
 
         self.fanout.pop(topic, None)
+        self.time_since_last_publish.pop(topic, None)
 
     async def leave(self, topic: str) -> None:
         # Note: the comments here are the near-exact algorithm description from the spec
@@ -514,10 +518,12 @@ def mesh_heartbeat(
 
     def fanout_heartbeat(self) -> None:
         # Note: the comments here are the exact pseudocode from the spec
-        for topic in self.fanout:
-            # Delete topic entry if it's not in `pubsub.peer_topics`
-            # or (TODO) if it's time-since-last-published > ttl
-            if topic not in self.pubsub.peer_topics:
+        for topic in list(self.fanout):
+            if (
+                topic not in self.pubsub.peer_topics
+                and self.time_since_last_publish.get(topic, 0) + self.time_to_live
+                < int(time.time())
+            ):
                 # Remove topic from fanout
                 del self.fanout[topic]
             else: