问django通道websocket中间件如何接收消息

EN

class _InterceptionMiddleware:
    def __init__(self, application_cls, scope):
        self.application = application_cls(scope)


    async def __call__(self, receive, send):
        self.downstream_send = send

        # create a Queue for the upstream consumer to read messages from
        self.upstream_receive_queue = asyncio.Queue()

        # create a Queue for the upstream consumer to write messages to
        self.upstream_send_queue = asyncio.Queue()

        # pipe messages being sent to the upstream consumer to your interceptor method
        receiver = await_many_dispatch([receive], self.my_receive_interceptor_method)

        # pipe messages being sent buy the upstream consumer to your interceptor method
        sender = await_many_dispatch(
            [self.upstream_send_queue.get],
            self.my_send_interceptor_method
        )

        # set up an asyncio task to handle these pipes
        receiver_task = asyncio.create_task(receiver)
        sender_task = asyncio.create_task(sender)

        # create an asyncio task for the upstream consumer
        upstream_task = asyncio.create_task(
            # pass the `get` and `put` methods of your upstream send and receive queues
            self.application(self.upstream_receive_queue.get, self.upstream_send_queue.put)
        )

        # await it all
        done, pending = await asyncio.wait(
            [upstream_task, receiver_task, sender_task],
            # if any of them fail stop
            return_when=asyncio.FIRST_COMPLETED
        )
        for task in [upstream_task, receiver_task, sender_task]:
            if not task.dont():
                # we need to cancel this task.
                task.cancel()
                try:
                    await task
                except CancelledError:
                    # we expect this error
                    pass


    async def my_receive_interceptor_method(self, msg):
        # your interception code
        await self.upstream_receive_queue.put(msg)


    async def my_send_interceptor_method(self, msg):
        # your interception code
        await self.downstream_send(msg)


def InterceptionMiddleware(application_cls):
    return functools.partial(_InterceptionMiddleware, application_cls)