diff --git a/transport/connection/__init__.py b/network/connection/__init__.py similarity index 100% rename from transport/connection/__init__.py rename to network/connection/__init__.py diff --git a/transport/connection/raw_connection.py b/network/connection/raw_connection.py similarity index 100% rename from transport/connection/raw_connection.py rename to network/connection/raw_connection.py diff --git a/transport/connection/raw_connection_interface.py b/network/connection/raw_connection_interface.py similarity index 100% rename from transport/connection/raw_connection_interface.py rename to network/connection/raw_connection_interface.py diff --git a/transport/stream/__init__.py b/network/stream/__init__.py similarity index 100% rename from transport/stream/__init__.py rename to network/stream/__init__.py diff --git a/network/stream/net_stream.py b/network/stream/net_stream.py new file mode 100644 index 00000000..0dca44dc --- /dev/null +++ b/network/stream/net_stream.py @@ -0,0 +1,42 @@ +import asyncio +from .net_stream_interface import INetStream + +class NetStream(INetStream): + + def __init__(self, muxed_stream): + self.muxed_stream = muxed_stream + + def get_protocol(self): + """ + :return: protocol id that stream runs on + """ + return self.protocol_id + + def set_protocol(self, protocol_id): + """ + :param protocol_id: protocol id that stream runs on + :return: true if successful + """ + self.protocol_id = protocol_id + + def read(self): + """ + read from stream + :return: bytes of input until EOF + """ + return self.muxed_stream.read() + + def write(self, bytes): + """ + write to stream + :return: number of bytes written + """ + return self.muxed_stream.write(bytes) + + def close(self): + """ + close stream + :return: true if successful + """ + self.muxed_stream.close() + return True diff --git a/transport/stream/stream_interface.py b/network/stream/net_stream_interface.py similarity index 97% rename from transport/stream/stream_interface.py rename to network/stream/net_stream_interface.py index 6d63a2f0..920e2faf 100644 --- a/transport/stream/stream_interface.py +++ b/network/stream/net_stream_interface.py @@ -1,6 +1,6 @@ from abc import ABC, abstractmethod -class IStream(ABC): +class INetStream(ABC): def __init__(self, peer_id, multi_addr, connection): self.peer_id = peer_id diff --git a/network/swarm.py b/network/swarm.py index edd4446a..6dcdcb44 100644 --- a/network/swarm.py +++ b/network/swarm.py @@ -1,13 +1,15 @@ +import uuid from .network_interface import INetwork -from muxer.mplex.muxed_connection import MuxedConn -from transport.connection.raw_connection import RawConnection +from .stream.net_stream import NetStream class Swarm(INetwork): - def __init__(self, my_peer_id, peerstore): + def __init__(self, my_peer_id, peerstore, upgrader): self.my_peer_id = my_peer_id self.peerstore = peerstore - self.connections = {} + self.upgrader = upgrader + self.connections = dict() + self.listeners = dict() def set_stream_handler(self, stream_handler): """ @@ -18,34 +20,71 @@ class Swarm(INetwork): def new_stream(self, peer_id, protocol_id): """ - Determine if a connection to peer_id already exists - If a connection to peer_id exists, then - c = existing connection, - otherwise c = new muxed connection to peer_id - s = c.open_stream(protocol_id) - return s - :param peer_id: peer_id of destination :param protocol_id: protocol id - :return: stream instance + :return: net stream instance """ - muxed_connection = None + muxed_conn = None if peer_id in self.connections: - muxed_connection = self.connections[peer_id] + """ + If muxed connection already exists for peer_id, + set muxed connection equal to + existing muxed connection + """ + muxed_conn = self.connections[peer_id] else: + # Get peer info from peer store addrs = self.peerstore.addrs(peer_id) - stream_ip = addrs.get_protocol_value("ip") - stream_port = addrs.get_protocol_value("port") - if len(addrs) > 0: - conn = RawConnection(stream_ip, stream_port) - muxed_connection = MuxedConnection(conn, True) - else: - raise Exception("No IP and port in addr") - return muxed_connection.open_stream(protocol_id, "", peer_id, addrs) + + # Transport dials peer (gets back a raw conn) + if not addrs: + raise SwarmException("No known addresses to peer") + first_addr = addrs[0] + raw_conn = self.transport.dial(first_addr) + + # Use upgrader to upgrade raw conn to muxed conn + muxed_conn = self.upgrader.upgrade_connection(raw_conn) + + # Store muxed connection in connections + self.connections[peer_id] = muxed_conn + + # Use muxed conn to open stream, which returns + # a muxed stream + stream_id = str(uuid.uuid4()) + muxed_stream = muxed_conn.open_stream(protocol_id, stream_id, peer_id, first_addr) + + # Create a net stream + net_stream = NetStream(muxed_stream) + + return net_stream def listen(self, *args): """ :param *args: one or many multiaddrs to start listening on :return: true if at least one success """ - pass + + # Create a closure C that takes in a multiaddr and + # returns a function object O that takes in a reader and writer. + # This function O looks up the stream handler + # for the given protocol, creates the net_stream + # for the listener and calls the stream handler function + # passing in the net_stream + + # For each multiaddr in args + # Check if a listener for multiaddr exists already + # If listener already exists, continue + # Otherwise, do the following: + # Pass multiaddr into C and get back function H + # listener = transport.create_listener(H) + # Call listener listen with the multiaddr + # Map multiaddr to listener + + return True + + def add_transport(self, transport): + # TODO: Support more than one transport + self.transport = transport + +class SwarmException(Exception): + pass diff --git a/transport/stream/stream.py b/transport/stream/stream.py deleted file mode 100644 index 34daf97f..00000000 --- a/transport/stream/stream.py +++ /dev/null @@ -1,58 +0,0 @@ -import asyncio -from .stream_interface import IStream - -class Stream(IStream): - - def __init__(self, peer_id, multi_addr, connection): - IStream.__init__(self, peer_id, multi_addr, connection) - self.peer_id = peer_id - - self.multi_addr = multi_addr - - self.stream_ip = multi_addr.get_protocol_value("ip4") - self.stream_port = multi_addr.get_protocol_value("tcp") - - self.reader = connection.reader - self.writer = connection.writer - - # TODO should construct protocol id from constructor - self.protocol_id = None - - def get_protocol(self): - """ - :return: protocol id that stream runs on - """ - return self.protocol_id - - def set_protocol(self, protocol_id): - """ - :param protocol_id: protocol id that stream runs on - :return: true if successful - """ - self.protocol_id = protocol_id - - def read(self): - """ - read from stream - :return: bytes of input - """ - return self.reader.read(-1) - - def write(self, _bytes): - """ - write to stream - :return: number of bytes written - """ - return self.write_to_stream(_bytes) - - async def write_to_stream(self, _bytes): - to_return = self.writer.write(_bytes) - await self.writer.drain() - return to_return - - def close(self): - """ - close stream - :return: true if successful - """ - self.writer.close() diff --git a/transport/upgrader.py b/transport/upgrader.py index 64014244..e70d4377 100644 --- a/transport/upgrader.py +++ b/transport/upgrader.py @@ -1,18 +1,24 @@ class TransportUpgrader(object): - def __init__(self, secOpt, muxerOpt): - self.sec = secOpt - self.muxer = muxerOpt + def __init__(self, secOpt, muxerOpt): + self.sec = secOpt + self.muxer = muxerOpt - def upgrade_listener(self, transport, listeners): - """ - upgrade multiaddr listeners to libp2p-transport listeners + def upgrade_listener(self, transport, listeners): + """ + upgrade multiaddr listeners to libp2p-transport listeners - """ - pass - - def upgrade_security(self): - pass + """ + pass + + def upgrade_security(self): + pass - def upgrade_muxer(self): - pass \ No newline at end of file + def upgrade_connection(self, conn): + """ + upgrade raw connection to muxed connection + """ + # For PoC, no security + # Default to mplex + pass + \ No newline at end of file