Fix cyclic import and lint

libp2p/host/basic_host.py

@@ -24,6 +24,9 @@ from .host_interface import IHost
 # telling it to listen on the given listen addresses.
 
 
+StreamHandlerFn = Callable[[INetStream], Coroutine[Any, Any, None]]
+
+
 class BasicHost(IHost):
 
     _network: Swarm
@@ -72,7 +75,7 @@ class BasicHost(IHost):
                 addrs.append(addr.encapsulate(p2p_part))
         return addrs
 
-    def set_stream_handler(self, protocol_id: str, stream_handler: Callable[[INetStream], Coroutine[Any, Any, None]]) -> bool:
+    def set_stream_handler(self, protocol_id: str, stream_handler: StreamHandlerFn) -> bool:
         """
         set stream handler for host
         :param protocol_id: protocol id used on stream

libp2p/host/host_interface.py

@@ -12,10 +12,11 @@ import multiaddr
 from libp2p.network.swarm import Swarm
 from libp2p.peer.id import ID
 from libp2p.peer.peerinfo import PeerInfo
-from libp2p.peer.peerstore import PeerStore
 
 from libp2p.network.stream.net_stream_interface import INetStream
-from libp2p.routing.kademlia.kademlia_peer_router import KadmeliaPeerRouter
+
+
+StreamHandlerFn = Callable[[INetStream], Coroutine[Any, Any, None]]
 
 
 class IHost(ABC):
@@ -46,7 +47,7 @@ class IHost(ABC):
         """
 
     @abstractmethod
-    def set_stream_handler(self, protocol_id: str, stream_handler: Callable[[INetStream], Coroutine[Any, Any, None]]) -> bool:
+    def set_stream_handler(self, protocol_id: str, stream_handler: StreamHandlerFn) -> bool:
         """
         set stream handler for host
         :param protocol_id: protocol id used on stream
@@ -57,7 +58,9 @@ class IHost(ABC):
     # protocol_id can be a list of protocol_ids
     # stream will decide which protocol_id to run on
     @abstractmethod
-    def new_stream(self, peer_id: ID, protocol_ids: Sequence[str]) -> Coroutine[Any, Any, INetStream]:
+    def new_stream(self,
+                   peer_id: ID,
+                   protocol_ids: Sequence[str]) -> Coroutine[Any, Any, INetStream]:
         """
         :param peer_id: peer_id that host is connecting
         :param protocol_ids: protocol ids that stream can run on

libp2p/network/network_interface.py

@@ -1,9 +1,13 @@
-from abc import ABC, abstractmethod
+from abc import (
+    ABC,
+    abstractmethod,
+)
 from typing import (
     Any,
     Callable,
     Coroutine,
     Sequence,
+    TYPE_CHECKING,
 )
 
 from multiaddr import Multiaddr
@@ -11,9 +15,14 @@ from multiaddr import Multiaddr
 from libp2p.peer.id import ID
 from libp2p.stream_muxer.muxed_connection_interface import IMuxedConn
 
-from .notifee_interface import INotifee
 from .stream.net_stream import NetStream
 
+if TYPE_CHECKING:
+    from .notifee_interface import INotifee
+
+
+StreamHandlerFn = Callable[[NetStream], Coroutine[Any, Any, None]]
+
 
 class INetwork(ABC):
 
@@ -34,7 +43,7 @@ class INetwork(ABC):
         """
 
     @abstractmethod
-    def set_stream_handler(self, protocol_id: str, stream_handler: Callable[[NetStream], Coroutine[Any, Any, None]]) -> bool:
+    def set_stream_handler(self, protocol_id: str, stream_handler: StreamHandlerFn) -> bool:
         """
         :param protocol_id: protocol id used on stream
         :param stream_handler: a stream handler instance
@@ -42,7 +51,9 @@ class INetwork(ABC):
         """
 
     @abstractmethod
-    def new_stream(self, peer_id: ID, protocol_ids: Sequence[str]) -> Coroutine[Any, Any, NetStream]:
+    def new_stream(self,
+                   peer_id: ID,
+                   protocol_ids: Sequence[str]) -> Coroutine[Any, Any, NetStream]:
         """
         :param peer_id: peer_id of destination
         :param protocol_ids: available protocol ids to use for stream
@@ -57,7 +68,7 @@ class INetwork(ABC):
         """
 
     @abstractmethod
-    def notify(self, notifee: INotifee) -> bool:
+    def notify(self, notifee: 'INotifee') -> bool:
         """
         :param notifee: object implementing Notifee interface
         :return: true if notifee registered successfully, false otherwise

libp2p/network/notifee_interface.py

@@ -1,52 +1,58 @@
-from abc import ABC, abstractmethod
+from abc import (
+    ABC,
+    abstractmethod,
+)
+from typing import TYPE_CHECKING
 
 from multiaddr import Multiaddr
 
-from libp2p.network.network_interface import INetwork
 from libp2p.stream_muxer.muxed_connection_interface import IMuxedConn
 
 from libp2p.network.stream.net_stream_interface import INetStream
 
+if TYPE_CHECKING:
+    from .network_interface import INetwork
+
 
 class INotifee(ABC):
 
     @abstractmethod
-    async def opened_stream(self, network: INetwork, stream: INetStream) -> None:
+    async def opened_stream(self, network: 'INetwork', stream: INetStream) -> None:
         """
         :param network: network the stream was opened on
         :param stream: stream that was opened
         """
 
     @abstractmethod
-    async def closed_stream(self, network: INetwork, stream: INetStream) -> None:
+    async def closed_stream(self, network: 'INetwork', stream: INetStream) -> None:
         """
         :param network: network the stream was closed on
         :param stream: stream that was closed
         """
 
     @abstractmethod
-    async def connected(self, network: INetwork, conn: IMuxedConn) -> None:
+    async def connected(self, network: 'INetwork', conn: IMuxedConn) -> None:
         """
         :param network: network the connection was opened on
         :param conn: connection that was opened
         """
 
     @abstractmethod
-    async def disconnected(self, network: INetwork, conn: IMuxedConn) -> None:
+    async def disconnected(self, network: 'INetwork', conn: IMuxedConn) -> None:
         """
         :param network: network the connection was closed on
         :param conn: connection that was closed
         """
 
     @abstractmethod
-    async def listen(self, network: INetwork, multiaddr: Multiaddr) -> None:
+    async def listen(self, network: 'INetwork', multiaddr: Multiaddr) -> None:
         """
         :param network: network the listener is listening on
         :param multiaddr: multiaddress listener is listening on
         """
 
     @abstractmethod
-    async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
+    async def listen_close(self, network: 'INetwork', multiaddr: Multiaddr) -> None:
         """
         :param network: network the connection was opened on
         :param multiaddr: multiaddress listener is no longer listening on

libp2p/network/swarm.py

@@ -31,6 +31,9 @@ from .connection.raw_connection import RawConnection
 from .stream.net_stream import NetStream
 
 
+StreamHandlerFn = Callable[[NetStream], Coroutine[Any, Any, None]]
+
+
 class Swarm(INetwork):
     # pylint: disable=too-many-instance-attributes,cell-var-from-loop,too-many-arguments
 
@@ -76,7 +79,7 @@ class Swarm(INetwork):
     def get_peer_id(self) -> ID:
         return self.self_id
 
-    def set_stream_handler(self, protocol_id: str, stream_handler: Callable[[NetStream], Coroutine[Any, Any, None]]) -> bool:
+    def set_stream_handler(self, protocol_id: str, stream_handler: StreamHandlerFn) -> bool:
         """
         :param protocol_id: protocol id used on stream
         :param stream_handler: a stream handler instance
@@ -150,7 +153,10 @@ class Swarm(INetwork):
         muxed_stream = await muxed_conn.open_stream(protocol_ids[0], multiaddr)
 
         # Perform protocol muxing to determine protocol to use
-        selected_protocol = await self.multiselect_client.select_one_of(list(protocol_ids), muxed_stream)
+        selected_protocol = await self.multiselect_client.select_one_of(
+            list(protocol_ids),
+            muxed_stream,
+        )
 
         # Create a net stream with the selected protocol
         net_stream = NetStream(muxed_stream)
@@ -180,7 +186,8 @@ class Swarm(INetwork):
             if str(multiaddr) in self.listeners:
                 return True
 
-            async def conn_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+            async def conn_handler(reader: asyncio.StreamReader,
+                                   writer: asyncio.StreamWriter) -> None:
                 # Read in first message (should be peer_id of initiator) and ack
                 peer_id = id_b58_decode((await reader.read(1024)).decode())
 

libp2p/peer/id.py

@@ -86,5 +86,5 @@ def id_from_private_key(key: RsaKey) -> ID:
 
 def digest(data: Union[str, bytes]) -> bytes:
     if not isinstance(data, bytes):
-        data_bytes = str(data).encode('utf8')
-    return hashlib.sha1(data_bytes).digest()
+        data = str(data).encode('utf8')
+    return hashlib.sha1(data).digest()

libp2p/peer/peerinfo.py

@@ -49,7 +49,7 @@ def info_from_p2p_addr(addr: multiaddr.Multiaddr) -> PeerInfo:
         addr = multiaddr.Multiaddr.join(*parts[:-1])
 
     peer_data = PeerData()
-    peer_data.add_addrs(addr)
+    peer_data.add_addrs([addr])
     peer_data.set_protocols([p.code for p in addr.protocols()])
 
     return PeerInfo(peer_id, peer_data)

libp2p/pubsub/gossipsub.py

@@ -79,6 +79,9 @@ class GossipSub(IPubsubRouter):
         self.mesh = {}
         self.fanout = {}
 
+        # Create peer --> protocol mapping
+        self.peers_to_protocol = {}
+
         # Create topic --> time since last publish map
         self.time_since_last_publish = {}
 
@@ -449,7 +452,9 @@ class GossipSub(IPubsubRouter):
         self.mcache.shift()
 
     @staticmethod
-    def select_from_minus(num_to_select: int, pool: Sequence[Any], minus: Sequence[Any]) -> List[Any]:
+    def select_from_minus(num_to_select: int,
+                          pool: Sequence[Any],
+                          minus: Sequence[Any]) -> List[Any]:
         """
         Select at most num_to_select subset of elements from the set (pool - minus) randomly.
         :param num_to_select: number of elements to randomly select
@@ -510,7 +515,7 @@ class GossipSub(IPubsubRouter):
         # Add all unknown message ids (ids that appear in ihave_msg but not in seen_seqnos) to list
         # of messages we want to request
         # FIXME: Update type of message ID
-        msg_ids_wanted = [
+        msg_ids_wanted: List[Any] = [
             msg_id
             for msg_id in ihave_msg.messageIDs
             if literal_eval(msg_id) not in seen_seqnos_and_peers

libp2p/pubsub/pubsub.py

@@ -6,6 +6,7 @@ from typing import (
     Dict,
     List,
     Tuple,
+    TYPE_CHECKING,
 )
 
 from lru import LRU
@@ -17,6 +18,8 @@ from libp2p.network.stream.net_stream_interface import INetStream
 
 from .pb import rpc_pb2
 from .pubsub_notifee import PubsubNotifee
+
+if TYPE_CHECKING:
     from .pubsub_router_interface import IPubsubRouter
 
 
@@ -31,17 +34,19 @@ class Pubsub:
     host: IHost
     my_id: ID
 
-    router: IPubsubRouter
+    router: 'IPubsubRouter'
 
-    peer_queue: asyncio.Queue[ID]
+    peer_queue: asyncio.Queue
 
     protocols: List[str]
 
-    incoming_msgs_from_peers: asyncio.Queue[rpc_pb2.Message]
-    outgoing_messages: asyncio.Queue[rpc_pb2.Message]
+    incoming_msgs_from_peers: asyncio.Queue
+    outgoing_messages: asyncio.Queue
 
     seen_messages: LRU
+
     my_topics: Dict[str, asyncio.Queue]
+
     # FIXME: Should be changed to `Dict[str, List[ID]]`
     peer_topics: Dict[str, List[str]]
     # FIXME: Should be changed to `Dict[ID, INetStream]`
@@ -52,7 +57,7 @@ class Pubsub:
 
     def __init__(self,
                  host: IHost,
-                 router: IPubsubRouter,
+                 router: 'IPubsubRouter',
                  my_id: ID,
                  cache_size: int = None) -> None:
         """
@@ -247,7 +252,7 @@ class Pubsub:
                 # for each topic
                 await self.my_topics[topic].put(publish_message)
 
-    async def subscribe(self, topic_id: str) -> asyncio.Queue[rpc_pb2.Message]:
+    async def subscribe(self, topic_id: str) -> asyncio.Queue:
         """
         Subscribe ourself to a topic
         :param topic_id: topic_id to subscribe to

libp2p/pubsub/pubsub_notifee.py

@@ -1,11 +1,7 @@
 import asyncio
-from typing import (
-    Sequence,
-)
 
 from multiaddr import Multiaddr
 
-from libp2p.peer.id import ID
 from libp2p.network.network_interface import INetwork
 from libp2p.network.notifee_interface import INotifee
 from libp2p.stream_muxer.muxed_connection_interface import IMuxedConn
@@ -16,9 +12,9 @@ from libp2p.network.stream.net_stream_interface import INetStream
 class PubsubNotifee(INotifee):
     # pylint: disable=too-many-instance-attributes, cell-var-from-loop
 
-    initiator_peers_queue: asyncio.Queue[ID]
+    initiator_peers_queue: asyncio.Queue
 
-    def __init__(self, initiator_peers_queue: asyncio.Queue[ID]) -> None:
+    def __init__(self, initiator_peers_queue: asyncio.Queue) -> None:
         """
         :param initiator_peers_queue: queue to add new peers to so that pubsub
         can process new peers after we connect to them

libp2p/pubsub/pubsub_router_interface.py

@@ -1,11 +1,14 @@
 from abc import ABC, abstractmethod
 from typing import (
     List,
+    TYPE_CHECKING,
 )
 
 from libp2p.peer.id import ID
 
 from .pb import rpc_pb2
+
+if TYPE_CHECKING:
     from .pubsub import Pubsub
 
 class IPubsubRouter(ABC):
@@ -17,7 +20,7 @@ class IPubsubRouter(ABC):
         """
 
     @abstractmethod
-    def attach(self, pubsub: Pubsub) -> None:
+    def attach(self, pubsub: 'Pubsub') -> None:
         """
         Attach is invoked by the PubSub constructor to attach the router to a
         freshly initialized PubSub instance.