import struct
import asyncio

def handle_msg(msg):
    idx = struct.unpack('I', msg[:4])[0]
    print('{0}: {1}'.format(idx, len(msg)))

@asyncio.coroutine
def subscribe_stuff():
    reader, writer = yield from asyncio.open_connection('127.0.0.1', 10000)

    hdrfmt = struct.Struct('>I')

    while True:
        try:
            hdr = yield from reader.readexactly(hdrfmt.size)
            packet_size = hdrfmt.unpack(hdr)[0]
            payload = yield from reader.readexactly(packet_size)
        except asyncio.IncompleteReadError as e:
            print('partial read {} / {}'.format(len(e.partial), e.expected))
            break

        handle_msg(payload)

loop = asyncio.get_event_loop()
loop.run_until_complete(subscribe_stuff())
#loop.run_forever()



@@ -23,16 +23,1352 @@
 syncio%0A%0A
+class SubscriberProtocol(asyncio.Protocol):%0A    hdrfmt = struct.Struct('%3EI')%0A%0A    def __init__(self, loop):%0A        self.loop = loop%0A%0A    def connection_made(self, transport):%0A        print('connection_made')%0A        self.transport = transport%0A        self.reset()%0A%0A    def connection_lost(self, exc):%0A        print('connection_lost')%0A        self.loop.stop()%0A%0A    def data_received(self, data):%0A        self.accum_buffer.extend(data)%0A%0A        while len(self.accum_buffer) %3E= self.packet_size:%0A            pkt = memoryview(self.accum_buffer)%5B:self.packet_size%5D%0A            self.accum_buffer = self.accum_buffer%5Bself.packet_size:%5D%0A            if self.wait_hdr:%0A                self.wait_hdr = False%0A                self.packet_size = self.hdrfmt.unpack(pkt)%5B0%5D%0A            else:%0A                self.wait_hdr = True%0A                self.packet_size = self.hdrfmt.size%0A                self.handle_msg(pkt)%0A%0A    def reset(self):%0A        self.wait_hdr = True%0A        self.packet_size = self.hdrfmt.size%0A        self.accum_buffer = bytearray()%0A%0A    def handle_msg(self, msg):%0A        idx = struct.unpack('I', msg%5B:4%5D)%5B0%5D%0A        print('%7B0%7D: %7B1%7D'.format(idx, len(msg)))%0A%0A@asyncio.coroutine%0Adef reconnect(loop):%0A    coro = loop.create_connection(lambda: SubscriberProtocol(loop), '127.0.0.1', 10000)%0A    transport, protocol = yield from coro%0A%0A
 def hand
@@ -2028,16 +2028,219 @@
 _loop()%0A
+#coro = loop.create_connection(lambda: SubscriberProtocol(loop), '127.0.0.1', 10000)%0A#transport, protocol = loop.run_until_complete(coro)%0Aasyncio.async(reconnect(loop))%0Aloop.run_forever()%0Aloop.close()%0A%0A#
 loop.run
@@ -2278,26 +2278,5 @@
 ())%0A
-#loop.run_forever()%0A%0A
 %0A

write subscriber as Protocol