(self, command: commands.ConsumeRpcsCommand)
Worker wishes for incoming RPC results to be listened for and processed
Worker wishes for incoming RPC results to be listened for and processed
Worker wishes for incoming RPC results to be listened for and processed
async def handle_consume_rpcs(self, command: commands.ConsumeRpcsCommand):
        """Worker wishes for incoming RPC results to be listened for and processed"""
        # Not all APIs will necessarily be served by the same transport, so group them
        # accordingly
        api_names_by_transport = self.transport_registry.get_rpc_transports(command.api_names)

        for rpc_transport, transport_api_names in api_names_by_transport.items():
            transport_apis = list(map(self.api_registry.get, transport_api_names))

            # fmt: off
            task = asyncio.ensure_future(queue_exception_checker(
                self._consume_rpcs_with_transport(rpc_transport=rpc_transport, apis=transport_apis),
                self.error_queue,
            ))
            # fmt: on
            self.consumer_tasks.add(task)
async def handle_consume_rpcs ( self , command : commands . ConsumeRpcsCommand ) : # Not all APIs will necessarily be served by the same transport, so group them # accordingly api_names_by_transport = self . transport_registry . get_rpc_transports ( command . api_names ) for rpc_transport , transport_api_names in api_names_by_transport . items ( ) : transport_apis = list ( map ( self . api_registry . get , transport_api_names ) ) # fmt: off task = asyncio . ensure_future ( queue_exception_checker ( self . _consume_rpcs_with_transport ( rpc_transport = rpc_transport , apis = transport_apis ) , self . error_queue , ) ) # fmt: on self . consumer_tasks . add ( task )