refactor for sprint

transport/connection/raw_connection.py

@@ -0,0 +1,16 @@
+import asyncio
+from .raw_connection_interface import IRawConnection
+
+class RawConnection(IRawConnection):
+
+    def __init__(self, ip, port):
+        self.conn_ip = ip
+        self.conn_port = port
+        self.reader, self.writer = self.open_connection()
+
+    async def open_connection(self):
+        """
+        opens a connection on self.ip and self.port
+        :return: a raw connection
+        """
+        return await asyncio.open_connection(self.conn_ip, self.conn_port)

transport/connection/raw_connection_interface.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+
+class IRawConnection(ABC):
+    """
+    A Raw Connection provides a Reader and a Writer
+    open_connection should return such a connection
+    """
+
+    @abstractmethod
+    async def open_connection(self):
+        """
+        opens a connection on ip and port
+        :return: a raw connection
+        """
+        pass

transport/listener_interface.py

@@ -0,0 +1,31 @@
+from abc import ABC, abstractmethod
+
+class IListener(ABC):
+
+    @abstractmethod
+    def listen(self, multiaddr):
+        """
+        put listener in listening mode and wait for incoming connections
+        :param multiaddr: multiaddr of peer
+        :return: return True if successful
+        """
+        pass
+
+    @abstractmethod
+    def get_addrs(self):
+        """
+        retrieve list of addresses the listener is listening on
+        :return: return list of addrs
+        """
+        pass
+
+    @abstractmethod
+    def close(self, options=None):
+        """
+        close the listener such that no more connections
+        can be open on this transport instance
+        :param options: optional object potential with timeout
+        a timeout value in ms that fires and destroy all connections
+        :return: return True if successful
+        """
+        pass

transport/stream/stream.py

@@ -0,0 +1,54 @@
+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)
+        self.peer_id = peer_id
+
+        stream_ip = multi_addr.get_protocol_value("ip4")
+        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()

transport/stream/stream_interface.py

@@ -0,0 +1,47 @@
+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

transport/tcp.py

@@ -5,24 +5,42 @@ from .listener_interface import IListener
 class TCP(ITransport):
 
     def __init__(self):
-        self.multiaddr = None
+        self.listener = self.Listener()
 
     class Listener(IListener):
 
+        def __init__(self, handler_function=None):
+            self.multiaddrs = []
+            self.server = None
+            self.handler = staticmethod(handler_function)
+
         def listen(self, multiaddr):
             """
             put listener in listening mode and wait for incoming connections
             :param multiaddr: multiaddr of peer
             :return: return True if successful
             """
-            pass
+            # TODO check for exceptions
+            _multiaddr = multiaddr
+            if "ipfs" in multiaddr.get_protocols():
+                # ipfs_id = multiaddr.get_ipfs_id()
+                _multiaddr = multiaddr.remove_protocol("ipfs")
+
+            self.multiaddrs.append(_multiaddr)
+            _multiaddr_dict = _multiaddr.to_dict()
+            _loop = asyncio.get_event_loop()
+            _coroutine = asyncio.start_server(self.handler, _multiaddr_dict.host,\
+                _multiaddr_dict.port, loop=_loop)
+            self.server = _loop.run_until_complete(_coroutine)
+            return True
 
         def get_addrs(self):
             """
             retrieve list of addresses the listener is listening on
             :return: return list of addrs
             """
-            pass
+            # TODO check if server is listening
+            return self.multiaddrs
 
         def close(self, options=None):
             """
@@ -32,16 +50,27 @@ class TCP(ITransport):
             a timeout value in ms that fires and destroy all connections
             :return: return True if successful
             """
-            pass
+            if self.server is None:
+                return False
+            self.server.close()
+            _loop = asyncio.get_event_loop()
+            _loop.run_until_complete(self.server.wait_closed())
+            _loop.close()
+            self.server = None
+            return True
 
     def dial(self, multiaddr, options=None):
         """
         dial a transport to peer listening on multiaddr
         :param multiaddr: multiaddr of peer
         :param options: optional object
-        :return: list of multiaddrs
+        :return: True if successful
         """
-        pass
+        _multiaddr_dict = multiaddr.to_dict()
+        reader, writer = await asyncio.open_connection(_multiaddr_dict.host,\
+            _multiaddr_dict.port)
+        return False
+        # TODO dial behavior not fully understood
 
     def create_listener(self, handler_function, options=None):
         """
@@ -51,4 +80,4 @@ class TCP(ITransport):
         that takes a connection as argument which implements interface-connection
         :return: a listener object that implements listener_interface.py
         """
-        pass
+        return self.Listener(handler_function)

transport/upgrader.py

@@ -0,0 +1,18 @@
+class TransportUpgrader(object):
+
+	def __init__(self, secOpt, muxerOpt):
+		self.sec = secOpt
+		self.muxer = muxerOpt
+
+	def upgrade_listener(self, transport, listeners):
+		"""
+		upgrade multiaddr listeners to libp2p-transport listeners
+
+		"""
+		pass
+		
+	def upgrade_security(self):
+		pass
+
+	def upgrade_muxer(self):
+		pass