autobahn_progressive_send_block.py

# USAGE: python minimal_example.py [receive | send]

from autobahn.asyncio.wamp import ApplicationSession
from autobahn.asyncio.wamp import ApplicationRunner
from autobahn.wamp.types import RegisterOptions
from autobahn.wamp import CallOptions

from autobahn import wamp

import io
import uuid
import functools

import asyncio

import txaio
txaio.use_asyncio()
txaio.start_logging(level='debug')

log = txaio.make_logger()

payload = b"0" * 2**26
CHUNK_SIZE = 2**15


def send_bytes_by_chunk(payload, details=None):
    #
    #
    # This is BLOCKING?
    #
    #
    if details.progress:
        for i in range(0, len(payload), CHUNK_SIZE):
            details.progress(payload[i:i+CHUNK_SIZE])
            print("send chunk!")
    else: return payload


class Sender(ApplicationSession):

    async def send(self):
        print("sending")

        uri = f'com.send.{str(uuid.uuid4())}'
        upload = functools.partial(send_bytes_by_chunk, payload, )
        registration = await self.register(
            upload,
            uri,
            options=RegisterOptions(details_arg='details')
        )


        print("ouch! next call is blocking.")
        await self.call('com.receive', uri)
        #
        #
        # We are BLOCKED here
        #
        #
        registration.unregister()

    @wamp.register('com.pouet')
    async def pouet(self):
        print("                                     pouet!")

    async def onJoin(self, details):
        await self.register(self)
        await self.send()


class Receiver(ApplicationSession):

    async def get_bytes_by_chunk(self, procedure_uri, *args, **kwargs):
        buffer = io.BytesIO()

        def on_progress(chunk):
            buffer.write(chunk)
            print("received chunk!")
            self.call('com.pouet')

        await self.call(
            procedure_uri,
            options=CallOptions(on_progress=on_progress),
            *args, **kwargs
        )
        payload = buffer.getvalue()
        buffer.close()

        return payload

    @wamp.register('com.receive')
    async def receive(self, source_uri):
        payload = await self.get_bytes_by_chunk(source_uri)
        print("receive payload. len: ", len(payload))

    async def onJoin(self, details):
        await self.register(self)

if __name__ == "__main__":
    import sys
    print('arg:', sys.argv[1])

    runner = ApplicationRunner(url=f"ws://localhost:8080/ws", realm='realm1')

    if sys.argv[1] == "send":
        runner.run(Sender, log_level='debug')
    elif sys.argv[1] == "receive":
        runner.run(Receiver, log_level='debug')