From 478cd033d5f0796e71cac2662fed55e7f2292f6a Mon Sep 17 00:00:00 2001 From: Alex Haynes Date: Sun, 11 Nov 2018 17:15:55 -0500 Subject: [PATCH 1/5] handle incoming, definition of send_message --- muxer/mplex/muxed_connection.py | 42 +++++++++++++++++++++++--- muxer/mplex/utils.py | 26 ++++++++++++++++ transport/connection/raw_connection.py | 4 +++ 3 files changed, 68 insertions(+), 4 deletions(-) create mode 100644 muxer/mplex/utils.py diff --git a/muxer/mplex/muxed_connection.py b/muxer/mplex/muxed_connection.py index b7caa620..c4f36585 100644 --- a/muxer/mplex/muxed_connection.py +++ b/muxer/mplex/muxed_connection.py @@ -1,3 +1,4 @@ +import asyncio from .muxed_connection_interface import IMuxedConn from transport.stream.Stream import Stream @@ -13,12 +14,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,13 +32,15 @@ class MuxedConn(IMuxedConn): """ pass - def open_stream(self, protocol_id, stream_name, peer_id, multi_addr): + 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 = Stream(peer_id, multi_addr, self) + self.streams[stream_id] = stream + self.buffers[stream_id] = bytearray() + return stream def accept_stream(self): @@ -42,3 +49,30 @@ class MuxedConn(IMuxedConn): :return: the accepted stream """ pass + + def send_message(self, header, data): + """ + sends a message over the connection + :param header: header to use + :param data: data to send in the message + :return: True if success + """ + pass + + async def handle_incoming(self): + data = bytearray() + while True: + chunk = self.raw_conn.reader.read(100) + if not chunk: + break + data += chunk + + # 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) diff --git a/muxer/mplex/utils.py b/muxer/mplex/utils.py new file mode 100644 index 00000000..c4696f85 --- /dev/null +++ b/muxer/mplex/utils.py @@ -0,0 +1,26 @@ +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): + shift = 0 + result = 0 + index = 0 + while True: + i = buff[index] + result |= (i & 0x7f) << shift + shift += 7 + if not i & 0x80: + break + index += 1 + + return result, index diff --git a/transport/connection/raw_connection.py b/transport/connection/raw_connection.py index efc3706a..4947b1c2 100644 --- a/transport/connection/raw_connection.py +++ b/transport/connection/raw_connection.py @@ -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 From 162f54af0cb963e9bbccd400cf83ee1add232481 Mon Sep 17 00:00:00 2001 From: Alex Haynes Date: Sun, 11 Nov 2018 17:38:11 -0500 Subject: [PATCH 2/5] send message done --- muxer/mplex/muxed_connection.py | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/muxer/mplex/muxed_connection.py b/muxer/mplex/muxed_connection.py index c4f36585..6c77c9bc 100644 --- a/muxer/mplex/muxed_connection.py +++ b/muxer/mplex/muxed_connection.py @@ -1,4 +1,5 @@ import asyncio +from .utils import encode_uvarint, decode_uvarint from .muxed_connection_interface import IMuxedConn from transport.stream.Stream import Stream @@ -50,14 +51,25 @@ class MuxedConn(IMuxedConn): """ pass - def send_message(self, header, data): + 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 """ - pass + # << 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): + to_return = self.raw_conn.writer.write(_bytes) + await self.raw_conn.writer.drain() + return to_return async def handle_incoming(self): data = bytearray() From e44e31e55b752bb48f3c97850e31d818433f7580 Mon Sep 17 00:00:00 2001 From: Alex Haynes Date: Sun, 11 Nov 2018 17:48:31 -0500 Subject: [PATCH 3/5] read buffer and write_to_stream return --- muxer/mplex/muxed_connection.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/muxer/mplex/muxed_connection.py b/muxer/mplex/muxed_connection.py index 6c77c9bc..98b4d54c 100644 --- a/muxer/mplex/muxed_connection.py +++ b/muxer/mplex/muxed_connection.py @@ -33,6 +33,11 @@ class MuxedConn(IMuxedConn): """ pass + 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 @@ -67,9 +72,9 @@ class MuxedConn(IMuxedConn): return self.write_to_stream(_bytes) async def write_to_stream(self, _bytes): - to_return = self.raw_conn.writer.write(_bytes) + self.raw_conn.writer.write(_bytes) await self.raw_conn.writer.drain() - return to_return + return len(_bytes) async def handle_incoming(self): data = bytearray() From cacbc6c11ac47c29d32259aa38732b75390de0b5 Mon Sep 17 00:00:00 2001 From: Alex Haynes Date: Sun, 11 Nov 2018 17:55:50 -0500 Subject: [PATCH 4/5] finished handle_incoming --- muxer/mplex/muxed_connection.py | 8 ++++++++ muxer/mplex/utils.py | 3 +-- 2 files changed, 9 insertions(+), 2 deletions(-) diff --git a/muxer/mplex/muxed_connection.py b/muxer/mplex/muxed_connection.py index 98b4d54c..e6d5819a 100644 --- a/muxer/mplex/muxed_connection.py +++ b/muxer/mplex/muxed_connection.py @@ -83,7 +83,15 @@ class MuxedConn(IMuxedConn): 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 diff --git a/muxer/mplex/utils.py b/muxer/mplex/utils.py index c4696f85..4e202416 100644 --- a/muxer/mplex/utils.py +++ b/muxer/mplex/utils.py @@ -11,10 +11,9 @@ def encode_uvarint(number): break return buf -def decode_uvarint(buff): +def decode_uvarint(buff, index): shift = 0 result = 0 - index = 0 while True: i = buff[index] result |= (i & 0x7f) << shift From ffc3fb059fbd4080b727aff31cc1dd4d402f9397 Mon Sep 17 00:00:00 2001 From: Alex Haynes Date: Sun, 11 Nov 2018 18:01:08 -0500 Subject: [PATCH 5/5] updated open stream --- muxer/mplex/muxed_connection.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/muxer/mplex/muxed_connection.py b/muxer/mplex/muxed_connection.py index e6d5819a..fdae239c 100644 --- a/muxer/mplex/muxed_connection.py +++ b/muxer/mplex/muxed_connection.py @@ -1,7 +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): """ @@ -43,7 +43,7 @@ class MuxedConn(IMuxedConn): creates a new muxed_stream :return: a new stream """ - stream = 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