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

Author: yashksaini-coder

examples/echo/echo.py

@@ -1,7 +1,6 @@
 import argparse
 import random
 import secrets
-import socket
 
 import multiaddr
 import trio
@@ -25,20 +24,14 @@
     info_from_p2p_addr,
 )
 from libp2p.utils.address_validation import (
+    find_free_port,
     get_available_interfaces,
 )
 
 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.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

libp2p/utils/address_validation.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import socket
+
 from multiaddr import Multiaddr
 
 try:
@@ -35,6 +37,13 @@ def _safe_get_network_addrs(ip_version: int) -> list[str]:
     return []
 
 
+def find_free_port() -> int:
+    """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]
+
+
 def _safe_expand(addr: Multiaddr, port: int | None = None) -> list[Multiaddr]:
     """
     Internal safe expansion wrapper. Returns a list of Multiaddr objects.
@@ -147,4 +156,5 @@ def is_non_loopback(ma: Multiaddr) -> bool:
     "get_available_interfaces",
     "get_optimal_binding_address",
     "expand_wildcard_address",
+    "find_free_port",
 ]