(self, rpc_message: RpcMessage, options: dict)
Emit a call to a remote procedure

        This only sends the request, it does not await any result (see RedisResultTransport)
Emit a call to a remote procedure
Emit a call to a remote procedure
async def call_rpc(self, rpc_message: RpcMessage, options: dict):
        """Emit a call to a remote procedure

        This only sends the request, it does not await any result (see RedisResultTransport)
        """
        queue_key = f"{rpc_message.api_name}:rpc_queue"
        expiry_key = f"rpc_expiry_key:{rpc_message.id}"
        logger.debug(
            LBullets(
                L("Enqueuing message {} in Redis list {}", Bold(rpc_message), Bold(queue_key)),
                items=dict(**rpc_message.get_metadata(), kwargs=rpc_message.get_kwargs()),
            )
        )

        start_time = time.time()
        for try_number in range(3, 0, -1):
            last_try = try_number == 1
            try:
                await self._call_rpc(rpc_message, queue_key, expiry_key)
                break
            except (PipelineError, ConnectionClosedError, ConnectionResetError) as e:
                if not last_try:
                    logger.debug(
                        f"Retrying sending of message. Will retry {try_number} more times. "
                        f"Error was {type(e).__name__}: {e}"
                    )
                    await asyncio.sleep(self.rpc_retry_delay)
                else:
                    raise

        logger.debug(
            L(
                "Enqueued message {} in Redis in {} stream {}",
                Bold(rpc_message),
                human_time(time.time() - start_time),
                Bold(queue_key),
            )
        )
async def call_rpc ( self , rpc_message : RpcMessage , options : dict ) : queue_key = f"{rpc_message.api_name}:rpc_queue" expiry_key = f"rpc_expiry_key:{rpc_message.id}" logger . debug ( LBullets ( L ( "Enqueuing message {} in Redis list {}" , Bold ( rpc_message ) , Bold ( queue_key ) ) , items = dict ( * * rpc_message . get_metadata ( ) , kwargs = rpc_message . get_kwargs ( ) ) , ) ) start_time = time . time ( ) for try_number in range ( 3 , 0 , - 1 ) : last_try = try_number == 1 try : await self . _call_rpc ( rpc_message , queue_key , expiry_key ) break except ( PipelineError , ConnectionClosedError , ConnectionResetError ) as e : if not last_try : logger . debug ( f"Retrying sending of message. Will retry {try_number} more times. " f"Error was {type(e).__name__}: {e}" ) await asyncio . sleep ( self . rpc_retry_delay ) else : raise logger . debug ( L ( "Enqueued message {} in Redis in {} stream {}" , Bold ( rpc_message ) , human_time ( time . time ( ) - start_time ) , Bold ( queue_key ) , ) )