refactor(app): 804 refactored find_free_port() in address_validation.py

examples/echo/echo.py

@@ -1,7 +1,6 @@
 import argparse
 import random
 import secrets
-import socket
 
 import multiaddr
 import trio
@@ -25,6 +24,7 @@ from libp2p.peer.peerinfo import (
     info_from_p2p_addr,
 )
 from libp2p.utils.address_validation import (
+    find_free_port,
     get_available_interfaces,
 )
 
@@ -32,13 +32,6 @@ PROTOCOL_ID = TProtocol("/echo/1.0.0")
 MAX_READ_LEN = 2**32 - 1
 
 
-def find_free_port():
-    """Find a free port on localhost."""
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("", 0))  # Bind to a free port provided by the OS
-        return s.getsockname()[1]
-
-
 async def _echo_stream_handler(stream: INetStream) -> None:
     try:
         peer_id = stream.muxed_conn.peer_id

examples/pubsub/pubsub.py

@@ -1,6 +1,5 @@
 import argparse
 import logging
-import socket
 
 import base58
 import multiaddr
@@ -31,6 +30,9 @@ from libp2p.stream_muxer.mplex.mplex import (
 from libp2p.tools.async_service.trio_service import (
     background_trio_service,
 )
+from libp2p.utils.address_validation import (
+    find_free_port,
+)
 
 # Configure logging
 logging.basicConfig(
@@ -77,13 +79,6 @@ async def publish_loop(pubsub, topic, termination_event):
             await trio.sleep(1)  # Avoid tight loop on error
 
 
-def find_free_port():
-    """Find a free port on localhost."""
-    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
-        s.bind(("", 0))  # Bind to a free port provided by the OS
-        return s.getsockname()[1]
-
-
 async def monitor_peer_topics(pubsub, nursery, termination_event):
     """
     Monitor for new topics that peers are subscribed to and