From 356cac02bf2bafa0d72ff642e12e7b20ba372eaf Mon Sep 17 00:00:00 2001 From: Stuckinaboot Date: Sun, 11 Nov 2018 16:04:57 -0500 Subject: [PATCH 1/4] Add mux stream --- network/stream/__init__.py | 0 network/stream/net_stream.py | 42 +++++++++++++++++++++++ network/stream/net_stream_interface.py | 47 ++++++++++++++++++++++++++ 3 files changed, 89 insertions(+) create mode 100644 network/stream/__init__.py create mode 100644 network/stream/net_stream.py create mode 100644 network/stream/net_stream_interface.py diff --git a/network/stream/__init__.py b/network/stream/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/network/stream/net_stream.py b/network/stream/net_stream.py new file mode 100644 index 0000000..0dca44d --- /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/network/stream/net_stream_interface.py b/network/stream/net_stream_interface.py new file mode 100644 index 0000000..920e2fa --- /dev/null +++ b/network/stream/net_stream_interface.py @@ -0,0 +1,47 @@ +from abc import ABC, abstractmethod + +class INetStream(ABC): + + def __init__(self, peer_id, multi_addr, connection): + self.peer_id = peer_id + self.multi_addr = multi_addr + self.connection = connection + + @abstractmethod + def get_protocol(self): + """ + :return: protocol id that stream runs on + """ + pass + + @abstractmethod + def set_protocol(self, protocol_id): + """ + :param protocol_id: protocol id that stream runs on + :return: true if successful + """ + pass + + @abstractmethod + def read(self): + """ + read from stream + :return: bytes of input + """ + pass + + @abstractmethod + def write(self, _bytes): + """ + write to stream + :return: number of bytes written + """ + pass + + @abstractmethod + def close(self): + """ + close stream + :return: true if successful + """ + pass From 6ba5793c1d21500efc47bd9682cdde35a018143a Mon Sep 17 00:00:00 2001 From: Stuckinaboot Date: Sun, 11 Nov 2018 16:09:37 -0500 Subject: [PATCH 2/4] Reorganize folders (stream and connection) --- {transport => network}/connection/__init__.py | 0 .../connection/raw_connection.py | 0 .../connection/raw_connection_interface.py | 0 network/swarm.py | 14 ++--- transport/stream/__init__.py | 0 transport/stream/stream.py | 58 ------------------- transport/stream/stream_interface.py | 47 --------------- transport/upgrader.py | 32 +++++----- 8 files changed, 25 insertions(+), 126 deletions(-) rename {transport => network}/connection/__init__.py (100%) rename {transport => network}/connection/raw_connection.py (100%) rename {transport => network}/connection/raw_connection_interface.py (100%) delete mode 100644 transport/stream/__init__.py delete mode 100644 transport/stream/stream.py delete mode 100644 transport/stream/stream_interface.py 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/network/swarm.py b/network/swarm.py index edd4446..e1ab955 100644 --- a/network/swarm.py +++ b/network/swarm.py @@ -4,7 +4,7 @@ from transport.connection.raw_connection import RawConnection 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 = {} @@ -18,19 +18,17 @@ 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 """ muxed_connection = None if peer_id in self.connections: + """ + If muxed connection already exists for peer_id, + set muxed connection equal to + existing muxed connection + """ muxed_connection = self.connections[peer_id] else: addrs = self.peerstore.addrs(peer_id) diff --git a/transport/stream/__init__.py b/transport/stream/__init__.py deleted file mode 100644 index e69de29..0000000 diff --git a/transport/stream/stream.py b/transport/stream/stream.py deleted file mode 100644 index 34daf97..0000000 --- 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/stream/stream_interface.py b/transport/stream/stream_interface.py deleted file mode 100644 index 6d63a2f..0000000 --- a/transport/stream/stream_interface.py +++ /dev/null @@ -1,47 +0,0 @@ -from abc import ABC, abstractmethod - -class IStream(ABC): - - def __init__(self, peer_id, multi_addr, connection): - self.peer_id = peer_id - self.multi_addr = multi_addr - self.connection = connection - - @abstractmethod - def get_protocol(self): - """ - :return: protocol id that stream runs on - """ - pass - - @abstractmethod - def set_protocol(self, protocol_id): - """ - :param protocol_id: protocol id that stream runs on - :return: true if successful - """ - pass - - @abstractmethod - def read(self): - """ - read from stream - :return: bytes of input - """ - pass - - @abstractmethod - def write(self, _bytes): - """ - write to stream - :return: number of bytes written - """ - pass - - @abstractmethod - def close(self): - """ - close stream - :return: true if successful - """ - pass diff --git a/transport/upgrader.py b/transport/upgrader.py index 6401424..e70d437 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 From 4851137ecefcf54a3b777370f138631ac27f4e4a Mon Sep 17 00:00:00 2001 From: Stuckinaboot Date: Sun, 11 Nov 2018 16:42:10 -0500 Subject: [PATCH 3/4] Implement new stream function --- network/swarm.py | 53 ++++++++++++++++++++++++++++++++++-------------- 1 file changed, 38 insertions(+), 15 deletions(-) diff --git a/network/swarm.py b/network/swarm.py index e1ab955..e05e3ab 100644 --- a/network/swarm.py +++ b/network/swarm.py @@ -1,6 +1,6 @@ +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): @@ -8,6 +8,7 @@ class Swarm(INetwork): self.my_peer_id = my_peer_id self.peerstore = peerstore self.connections = {} + self.upgrader = upgrader def set_stream_handler(self, stream_handler): """ @@ -20,26 +21,41 @@ class Swarm(INetwork): """ :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: """ If muxed connection already exists for peer_id, - set muxed connection equal to + set muxed connection equal to existing muxed connection - """ - muxed_connection = self.connections[peer_id] + """ + 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): """ @@ -47,3 +63,10 @@ class Swarm(INetwork): :return: true if at least one success """ pass + + def add_transport(self, transport): + # TODO: Support more than one transport + self.transport = transport + +class SwarmException(Exception): + pass From f6f2f6725d28f1eab4a1e8036c5baf24e598ef16 Mon Sep 17 00:00:00 2001 From: Stuckinaboot Date: Sun, 11 Nov 2018 17:10:37 -0500 Subject: [PATCH 4/4] Describe listen stub --- network/swarm.py | 22 ++++++++++++++++++++-- 1 file changed, 20 insertions(+), 2 deletions(-) diff --git a/network/swarm.py b/network/swarm.py index e05e3ab..6dcdcb4 100644 --- a/network/swarm.py +++ b/network/swarm.py @@ -7,8 +7,9 @@ class Swarm(INetwork): 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): """ @@ -62,7 +63,24 @@ class Swarm(INetwork): :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