todo: parse listen_addrs to set transport in SwarmBuilder (#616)

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 @@ from typing import (
     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
+    if listen_addrs is None:
-    transport = TCP()
+        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:

newsfragments/616.feature.rst

@@ -0,0 +1 @@
+Allow passing `listen_addrs` to `new_swarm` to customize swarm listening behavior.

tests/core/network/test_swarm.py

@@ -7,12 +7,18 @@ from trio.testing import (
     wait_all_tasks_blocked,
 )
 
+from libp2p import (
+    new_swarm,
+)
 from libp2p.network.exceptions import (
     SwarmException,
 )
 from libp2p.tools.utils import (
     connect_swarm,
 )
+from libp2p.transport.tcp.tcp import (
+    TCP,
+)
 from tests.utils.factories import (
     SwarmFactory,
 )
@@ -156,3 +162,20 @@ async def test_swarm_multiaddr(security_protocol):
 
         swarms[0].peerstore.add_addrs(swarms[1].get_peer_id(), addrs + addrs, 10000)
         await swarms[0].dial_peer(swarms[1].get_peer_id())
+
+
+def test_new_swarm_defaults_to_tcp():
+    swarm = new_swarm()
+    assert isinstance(swarm.transport, TCP)
+
+
+def test_new_swarm_tcp_multiaddr_supported():
+    addr = Multiaddr("/ip4/127.0.0.1/tcp/9999")
+    swarm = new_swarm(listen_addrs=[addr])
+    assert isinstance(swarm.transport, TCP)
+
+
+def test_new_swarm_quic_multiaddr_raises():
+    addr = Multiaddr("/ip4/127.0.0.1/udp/9999/quic")
+    with pytest.raises(ValueError, match="QUIC not yet supported"):
+        new_swarm(listen_addrs=[addr])