diff --git a/libp2p/host/basic_host.py b/libp2p/host/basic_host.py index d5e5fd1..e4a54cf 100644 --- a/libp2p/host/basic_host.py +++ b/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 @@ -31,7 +34,7 @@ class BasicHost(IHost): peerstore: PeerStore # default options constructor - def __init__(self, network: Swarm, router: KadmeliaPeerRouter=None) -> None: + def __init__(self, network: Swarm, router: KadmeliaPeerRouter = None) -> None: self._network = network self._router = router self.peerstore = self._network.peerstore @@ -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 diff --git a/libp2p/host/host_interface.py b/libp2p/host/host_interface.py index d158a9b..bf4fe5f 100644 --- a/libp2p/host/host_interface.py +++ b/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 diff --git a/libp2p/network/network_interface.py b/libp2p/network/network_interface.py index d3dffde..0a4a8eb 100644 --- a/libp2p/network/network_interface.py +++ b/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 diff --git a/libp2p/network/notifee_interface.py b/libp2p/network/notifee_interface.py index 00ca157..65ad448 100644 --- a/libp2p/network/notifee_interface.py +++ b/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 diff --git a/libp2p/network/swarm.py b/libp2p/network/swarm.py index e457981..68b1892 100644 --- a/libp2p/network/swarm.py +++ b/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()) diff --git a/libp2p/peer/id.py b/libp2p/peer/id.py index 461787f..3a4c4a3 100644 --- a/libp2p/peer/id.py +++ b/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() diff --git a/libp2p/peer/peerinfo.py b/libp2p/peer/peerinfo.py index 6271f18..6ccd023 100644 --- a/libp2p/peer/peerinfo.py +++ b/libp2p/peer/peerinfo.py @@ -17,7 +17,7 @@ class PeerInfo: peer_id: ID addrs: List[multiaddr.Multiaddr] - def __init__(self, peer_id: ID, peer_data: PeerData=None) -> None: + def __init__(self, peer_id: ID, peer_data: PeerData = None) -> None: self.peer_id = peer_id self.addrs = peer_data.get_addrs() if peer_data else None @@ -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) diff --git a/libp2p/pubsub/gossipsub.py b/libp2p/pubsub/gossipsub.py index e4b105c..c778d66 100644 --- a/libp2p/pubsub/gossipsub.py +++ b/libp2p/pubsub/gossipsub.py @@ -60,9 +60,9 @@ class GossipSub(IPubsubRouter): degree_low: int, degree_high: int, time_to_live: int, - gossip_window: int=3, - gossip_history: int=5, - heartbeat_interval: int=120) -> None: + gossip_window: int = 3, + gossip_history: int = 5, + heartbeat_interval: int = 120) -> None: # pylint: disable=too-many-arguments self.protocols = list(protocols) self.pubsub = None @@ -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 diff --git a/libp2p/pubsub/pubsub.py b/libp2p/pubsub/pubsub.py index f7fab10..5e86015 100644 --- a/libp2p/pubsub/pubsub.py +++ b/libp2p/pubsub/pubsub.py @@ -6,6 +6,7 @@ from typing import ( Dict, List, Tuple, + TYPE_CHECKING, ) from lru import LRU @@ -17,7 +18,9 @@ from libp2p.network.stream.net_stream_interface import INetStream from .pb import rpc_pb2 from .pubsub_notifee import PubsubNotifee -from .pubsub_router_interface import IPubsubRouter + +if TYPE_CHECKING: + from .pubsub_router_interface import IPubsubRouter def get_msg_id(msg: rpc_pb2.Message) -> Tuple[bytes, bytes]: @@ -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 diff --git a/libp2p/pubsub/pubsub_notifee.py b/libp2p/pubsub/pubsub_notifee.py index 880c06d..cdaec4b 100644 --- a/libp2p/pubsub/pubsub_notifee.py +++ b/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 diff --git a/libp2p/pubsub/pubsub_router_interface.py b/libp2p/pubsub/pubsub_router_interface.py index ce49b4f..0cc6c12 100644 --- a/libp2p/pubsub/pubsub_router_interface.py +++ b/libp2p/pubsub/pubsub_router_interface.py @@ -1,12 +1,15 @@ from abc import ABC, abstractmethod from typing import ( List, + TYPE_CHECKING, ) from libp2p.peer.id import ID from .pb import rpc_pb2 -from .pubsub import Pubsub + +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. diff --git a/libp2p/routing/interfaces.py b/libp2p/routing/interfaces.py index 5f7dfc2..de2ec7a 100644 --- a/libp2p/routing/interfaces.py +++ b/libp2p/routing/interfaces.py @@ -13,7 +13,7 @@ from libp2p.peer.peerinfo import PeerInfo class IContentRouting(ABC): @abstractmethod - def provide(self, cid: bytes, announce: bool=True) -> None: + def provide(self, cid: bytes, announce: bool = True) -> None: """ Provide adds the given cid to the content routing system. If announce is True, it also announces it, otherwise it is just kept in the local diff --git a/libp2p/routing/kademlia/kademlia_content_router.py b/libp2p/routing/kademlia/kademlia_content_router.py index e5f31d8..468ca3a 100644 --- a/libp2p/routing/kademlia/kademlia_content_router.py +++ b/libp2p/routing/kademlia/kademlia_content_router.py @@ -8,7 +8,7 @@ from libp2p.routing.interfaces import IContentRouting class KadmeliaContentRouter(IContentRouting): - def provide(self, cid: bytes, announce: bool=True) -> None: + def provide(self, cid: bytes, announce: bool = True) -> None: """ Provide adds the given cid to the content routing system. If announce is True, it also announces it, otherwise it is just kept in the local