Merge branch 'master' of https://github.com/zixuanzh/py-libp2p into mux-conn

This commit is contained in:
Alex Haynes 2018-11-11 17:35:33 -05:00
commit d0e6d00b15
9 changed files with 124 additions and 95 deletions

View 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

View File

@ -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

View File

@ -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
"""
# 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

View File

@ -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()

View File

@ -14,5 +14,11 @@ class TransportUpgrader(object):
def upgrade_security(self):
pass
def upgrade_muxer(self):
def upgrade_connection(self, conn):
"""
upgrade raw connection to muxed connection
"""
# For PoC, no security
# Default to mplex
pass