ZX
GitHub
commit
6708cde8dd
42
network/stream/net_stream.py
Normal file
42
network/stream/net_stream.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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
|
||||
def upgrade_connection(self, conn):
|
||||
"""
|
||||
upgrade raw connection to muxed connection
|
||||
"""
|
||||
# For PoC, no security
|
||||
# Default to mplex
|
||||
pass
|
||||
|
||||
Loading…
Reference in New Issue
Block a user