py-libp2p/network/swarm.py

159 lines
5.6 KiB
Python
Raw Normal View History

from protocol_muxer.multiselect_client import MultiselectClient
from protocol_muxer.multiselect import Multiselect
2018-10-15 13:52:25 +08:00
from .network_interface import INetwork
2018-11-12 05:42:10 +08:00
from .stream.net_stream import NetStream
2018-11-12 09:29:17 +08:00
from .connection.raw_connection import RawConnection
2018-10-15 01:47:06 +08:00
2018-10-15 13:52:25 +08:00
class Swarm(INetwork):
# pylint: disable=too-many-instance-attributes, cell-var-from-loop
2018-10-15 13:52:25 +08:00
def __init__(self, peer_id, peerstore, upgrader):
self.self_id = peer_id
2018-11-12 01:36:15 +08:00
self.peerstore = peerstore
2018-11-12 05:42:10 +08:00
self.upgrader = upgrader
2018-11-12 06:10:37 +08:00
self.connections = dict()
self.listeners = dict()
2018-11-12 09:29:17 +08:00
self.stream_handlers = dict()
2018-11-13 02:02:49 +08:00
self.transport = None
2018-10-15 13:52:25 +08:00
# Protocol muxing
self.multiselect = Multiselect()
self.multiselect_client = MultiselectClient()
2018-11-19 00:22:56 +08:00
def get_peer_id(self):
return self.self_id
2018-11-19 00:22:56 +08:00
2018-11-12 09:29:17 +08:00
def set_stream_handler(self, protocol_id, stream_handler):
2018-10-15 13:52:25 +08:00
"""
2018-11-12 09:29:17 +08:00
:param protocol_id: protocol id used on stream
2018-10-15 13:52:25 +08:00
:param stream_handler: a stream handler instance
:return: true if successful
"""
self.multiselect.add_handler(protocol_id, stream_handler)
return True
2018-10-15 13:52:25 +08:00
async def dial_peer(self, peer_id):
2018-11-01 05:40:01 +08:00
"""
dial_peer try to create a connection to peer_id
:param peer_id: peer if we want to dial
:raises SwarmException: raised when no address if found for peer_id
:return: muxed connection
2018-11-01 05:40:01 +08:00
"""
2018-11-13 00:00:43 +08:00
# Get peer info from peer store
addrs = self.peerstore.addrs(peer_id)
if not addrs:
raise SwarmException("No known addresses to peer")
# TODO: define logic to choose which address to use, or try them all ?
2018-11-13 00:00:43 +08:00
multiaddr = addrs[0]
2018-11-01 05:40:01 +08:00
if peer_id in self.connections:
2018-11-13 02:02:49 +08:00
# If muxed connection already exists for peer_id,
# set muxed connection equal to existing muxed connection
2018-11-12 05:42:10 +08:00
muxed_conn = self.connections[peer_id]
2018-11-01 05:40:01 +08:00
else:
2018-11-12 05:42:10 +08:00
# Transport dials peer (gets back a raw conn)
raw_conn = await self.transport.dial(multiaddr)
2018-11-12 05:42:10 +08:00
# Use upgrader to upgrade raw conn to muxed conn
2018-11-12 08:37:07 +08:00
muxed_conn = self.upgrader.upgrade_connection(raw_conn, True)
2018-11-12 05:42:10 +08:00
# Store muxed connection in connections
self.connections[peer_id] = muxed_conn
return muxed_conn
async def new_stream(self, peer_id, protocol_ids):
"""
:param peer_id: peer_id of destination
:param protocol_id: protocol id
:return: net stream instance
"""
# Get peer info from peer store
addrs = self.peerstore.addrs(peer_id)
if not addrs:
raise SwarmException("No known addresses to peer")
multiaddr = addrs[0]
muxed_conn = await self.dial_peer(peer_id)
2018-11-12 05:42:10 +08:00
# Use muxed conn to open stream, which returns
# a muxed stream
# TODO: Remove protocol id from being passed into muxed_conn
muxed_stream = await muxed_conn.open_stream(protocol_ids[0], peer_id, multiaddr)
# Perform protocol muxing to determine protocol to use
selected_protocol = await self.multiselect_client.select_one_of(protocol_ids, muxed_stream)
2018-11-12 05:42:10 +08:00
# Create a net stream with the selected protocol
2018-11-12 05:42:10 +08:00
net_stream = NetStream(muxed_stream)
net_stream.set_protocol(selected_protocol)
2018-11-12 05:42:10 +08:00
return net_stream
2018-10-22 01:51:55 +08:00
2018-11-13 00:00:43 +08:00
async def listen(self, *args):
2018-10-22 01:51:55 +08:00
"""
:param *args: one or many multiaddrs to start listening on
:return: true if at least one success
2018-11-13 02:02:49 +08:00
2018-11-13 00:00:43 +08:00
For each multiaddr in args
Check if a listener for multiaddr exists already
If listener already exists, continue
Otherwise:
Capture multiaddr in conn handler
Have conn handler delegate to stream handler
Call listener listen with the multiaddr
Map multiaddr to listener
"""
for multiaddr in args:
if str(multiaddr) in self.listeners:
2018-11-12 09:29:17 +08:00
return True
2018-11-13 00:00:43 +08:00
async def conn_handler(reader, writer):
2018-11-13 02:02:49 +08:00
# Upgrade reader/write to a net_stream and pass \
# to appropriate stream handler (using multiaddr)
raw_conn = RawConnection(multiaddr.value_for_protocol('ip4'),
multiaddr.value_for_protocol('tcp'), reader, writer)
2018-11-12 09:41:06 +08:00
muxed_conn = self.upgrader.upgrade_connection(raw_conn, False)
2018-11-12 09:29:17 +08:00
# TODO: Remove protocol id from muxed_conn accept stream or
# move protocol muxing into accept_stream
muxed_stream, _, _ = await muxed_conn.accept_stream()
# Perform protocol muxing to determine protocol to use
selected_protocol, handler = await self.multiselect.negotiate(muxed_stream)
2018-11-12 09:29:17 +08:00
net_stream = NetStream(muxed_stream)
net_stream.set_protocol(selected_protocol)
2018-11-12 09:29:17 +08:00
# Give to stream handler
# TODO: handle case when stream handler is set
2018-11-13 00:00:43 +08:00
# TODO: handle case of multiple protocols over same raw connection
await handler(net_stream)
2018-11-12 09:29:17 +08:00
try:
# Success
listener = self.transport.create_listener(conn_handler)
self.listeners[str(multiaddr)] = listener
2018-11-13 00:00:43 +08:00
await listener.listen(multiaddr)
2018-11-12 09:29:17 +08:00
return True
except IOError:
# Failed. Continue looping.
2018-11-13 00:00:43 +08:00
print("Failed to connect to: " + str(multiaddr))
2018-11-12 09:29:17 +08:00
# No multiaddr succeeded
return False
2018-11-12 05:42:10 +08:00
def add_transport(self, transport):
# TODO: Support more than one transport
self.transport = transport
2018-11-12 05:42:10 +08:00
class SwarmException(Exception):
pass