StarMirror/py-libp2p
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge branch 'master' into muxed-stream

Browse Source
This commit is contained in:
ZX 2018-11-11 19:20:10 -05:00 committed by GitHub
commit 8bb1dab646
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 217 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

68
muxer/mplex/muxed_connection.py
View File

@ -1,5 +1,7 @@
import asyncio
from .utils import encode_uvarint, decode_uvarint
from .muxed_connection_interface import IMuxedConn
from transport.stream.Stream import Stream
from .muxed_stream import MuxedStream
class MuxedConn(IMuxedConn):
"""
@ -13,12 +15,16 @@ class MuxedConn(IMuxedConn):
"""
self.raw_conn = conn
self.initiator = initiator
self.buffers = {}
self.streams = {}
self.add_incoming_task()
def close(self):
"""
close the stream muxer and underlying raw connection
"""
pass
self.raw_conn.close()
def is_closed(self):
"""
@ -27,12 +33,20 @@ class MuxedConn(IMuxedConn):
"""
pass
def open_stream(self, protocol_id, stream_name, peer_id, multi_addr):
def read_buffer(self, stream_id):
data = self.buffers[stream_id]
self.buffers[stream_id] = bytearray()
return data
def open_stream(self, protocol_id, stream_id, peer_id, multi_addr):
"""
creates a new muxed_stream
:return: a new stream
"""
return Stream(peer_id, multi_addr, self)
stream = MuxedStream(peer_id, multi_addr, self)
self.streams[stream_id] = stream
self.buffers[stream_id] = bytearray()
return stream
def accept_stream(self):
"""
@ -40,3 +54,49 @@ class MuxedConn(IMuxedConn):
:return: the accepted stream
"""
pass
def send_message(self, flag, data, stream_id):
"""
sends a message over the connection
:param header: header to use
:param data: data to send in the message
:param stream_id: stream the message is in
:return: True if success
"""
# << by 3, then or with flag
header = (stream_id << 3) | flag
header = encode_uvarint(header)
data_length = encode_uvarint(len(data))
_bytes = header + data_length + data
return self.write_to_stream(_bytes)
async def write_to_stream(self, _bytes):
self.raw_conn.writer.write(_bytes)
await self.raw_conn.writer.drain()
return len(_bytes)
async def handle_incoming(self):
data = bytearray()
while True:
chunk = self.raw_conn.reader.read(100)
if not chunk:
break
data += chunk
header, end_index = decode_uvarint(data, 0)
length, end_index = decode_uvarint(data, end_index)
message = data[end_index, end_index + length]
# Deal with other types of messages
flag = header & 0x07
stream_id = header >> 3
self.buffers[stream_id] = self.buffers[stream_id] + message
# Read header
# Read message length
# Read message into corresponding buffer
def add_incoming_task(self):
loop = asyncio.get_event_loop()
handle_incoming_task = loop.create_task(self.handle_incoming())
handle_incoming_task.add_done_callback(self.add_incoming_task)

25
muxer/mplex/utils.py Normal file
View File

@ -0,0 +1,25 @@
def encode_uvarint(number):
"""Pack `number` into varint bytes"""
buf = b''
while True:
towrite = number & 0x7f
number >>= 7
if number:
buf += bytes((towrite | 0x80, ))
else:
buf += bytes((towrite, ))
break
return buf
def decode_uvarint(buff, index):
shift = 0
result = 0
while True:
i = buff[index]
result |= (i & 0x7f) << shift
shift += 7
if not i & 0x80:
break
index += 1
return result, index

0
transport/connection/__init__.py → network/connection/__init__.py
View File

4
transport/connection/raw_connection.py → network/connection/raw_connection.py
View File

@ -9,6 +9,10 @@ class RawConnection(IRawConnection):
self.reader = reader
self.writer = writer
def close(self):
self.writer.close()
# def __init__(self, ip, port):
# self.conn_ip = ip
# self.conn_port = port

0
transport/connection/raw_connection_interface.py → network/connection/raw_connection_interface.py
View File

0
transport/stream/__init__.py → network/stream/__init__.py
View File

42
network/stream/net_stream.py Normal file
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

2
transport/stream/stream_interface.py → network/stream/net_stream_interface.py
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

85
network/swarm.py
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
"""
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

58
transport/stream/stream.py
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()

30
transport/upgrader.py
View File

@ -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
"""
pass
def upgrade_security(self):
pass
def upgrade_security(self):
pass
def upgrade_connection(self, conn):
"""
upgrade raw connection to muxed connection
"""
# For PoC, no security
# Default to mplex
pass
def upgrade_muxer(self):
pass