(self, apis: Sequence[Api])
return await retry_on_redis_connection_failure(
            fn=self._consume_rpcs,
            args=[apis],
            retry_delay=self.consumption_restart_delay,
            action="consuming RPCs",
        )
Consume RPCs for the given APIs
Consume RPCs for the given APIs
Consume RPCs for the given APIs
async def consume_rpcs(self, apis: Sequence[Api]) -> Sequence[RpcMessage]:
        """Consume RPCs for the given APIs"""
        if self._closed:
            # Triggered during shutdown
            raise TransportIsClosed("Transport is closed. Cannot consume RPCs")

        return await retry_on_redis_connection_failure(
            fn=self._consume_rpcs,
            args=[apis],
            retry_delay=self.consumption_restart_delay,
            action="consuming RPCs",
        )
async def consume_rpcs ( self , apis : Sequence [ Api ] ) -> Sequence [ RpcMessage ] : if self . _closed : # Triggered during shutdown raise TransportIsClosed ( "Transport is closed. Cannot consume RPCs" ) return await retry_on_redis_connection_failure ( fn = self . _consume_rpcs , args = [ apis ] , retry_delay = self . consumption_restart_delay , action = "consuming RPCs" , )