(
        self, api_name: str, name: str, kwargs: dict = frozendict(), options: dict = frozendict()
    )
return result_message.result
Perform an RPC call

        Call an RPC and return the result.
Perform an RPC call
Perform an RPC call
async def call_rpc_remote(
        self, api_name: str, name: str, kwargs: dict = frozendict(), options: dict = frozendict()
    ):
        """ Perform an RPC call

        Call an RPC and return the result.
        """
        kwargs = deform_to_bus(kwargs)
        rpc_message = RpcMessage(api_name=api_name, procedure_name=name, kwargs=kwargs)
        validate_event_or_rpc_name(api_name, "rpc", name)

        logger.info("📞  Calling remote RPC {}.{}".format(Bold(api_name), Bold(name)))

        start_time = time.time()

        validate_outgoing(self.config, self.schema, rpc_message)

        await self.hook_registry.execute("before_rpc_call", rpc_message=rpc_message)

        result_queue = InternalQueue()

        # Send the RPC
        await self.producer.send(
            commands.CallRpcCommand(message=rpc_message, options=options)
        ).wait()

        # Start a listener which will wait for results
        await self.producer.send(
            commands.ReceiveResultCommand(
                message=rpc_message, destination_queue=result_queue, options=options
            )
        ).wait()

        # Wait for the result from the listener we started.
        # The RpcResultDock will handle timeouts
        result = await bail_on_error(self.error_queue, result_queue.get())

        call_time = time.time() - start_time

        try:
            if isinstance(result, Exception):
                raise result
        except asyncio.TimeoutError:
            raise LightbusTimeout(
                f"Timeout when calling RPC {rpc_message.canonical_name} after waiting for {human_time(call_time)}. "
                f"It is possible no Lightbus process is serving this API, or perhaps it is taking "
                f"too long to process the request. In which case consider raising the 'rpc_timeout' "
                f"config option."
            ) from None
        else:
            assert isinstance(result, ResultMessage)
            result_message = result

        await self.hook_registry.execute(
            "after_rpc_call", rpc_message=rpc_message, result_message=result_message
        )

        if not result_message.error:
            logger.info(
                L(
                    "🏁  Remote call of {} completed in {}",
                    Bold(rpc_message.canonical_name),
                    human_time(call_time),
                )
            )
        else:
            logger.warning(
                L(
                    "⚡ Error during remote call of RPC {}. Took {}: {}",
                    Bold(rpc_message.canonical_name),
                    human_time(call_time),
                    result_message.result,
                )
            )
            raise LightbusWorkerError(
                "Error while calling {}: {}\nRemote stack trace:\n{}".format(
                    rpc_message.canonical_name, result_message.result, result_message.trace
                )
            )

        validate_incoming(self.config, self.schema, result_message)

        return result_message.result
async def call_rpc_remote ( self , api_name : str , name : str , kwargs : dict = frozendict ( ) , options : dict = frozendict ( ) ) : kwargs = deform_to_bus ( kwargs ) rpc_message = RpcMessage ( api_name = api_name , procedure_name = name , kwargs = kwargs ) validate_event_or_rpc_name ( api_name , "rpc" , name ) logger . info ( "📞  Calling remote RPC {}.{}".fo r mat(Bo l d(ap i _name),  B o d(na m e))) start_time = time . time ( ) validate_outgoing ( self . config , self . schema , rpc_message ) await self . hook_registry . execute ( "before_rpc_call" , rpc_message = rpc_message ) result_queue = InternalQueue ( ) # Send the RPC await self . producer . send ( commands . CallRpcCommand ( message = rpc_message , options = options ) ) . wait ( ) # Start a listener which will wait for results await self . producer . send ( commands . ReceiveResultCommand ( message = rpc_message , destination_queue = result_queue , options = options ) ) . wait ( ) # Wait for the result from the listener we started. # The RpcResultDock will handle timeouts result = await bail_on_error ( self . error_queue , result_queue . get ( ) ) call_time = time . time ( ) - start_time try : if isinstance ( result , Exception ) : raise result except asyncio . TimeoutError : raise LightbusTimeout ( f"Timeout when calling RPC {rpc_message.canonical_name} after waiting for {human_time(call_time)}. " f"It is possible no Lightbus process is serving this API, or perhaps it is taking " f"too long to process the request. In which case consider raising the 'rpc_timeout' " f"config option." ) from None else : assert isinstance ( result , ResultMessage ) result_message = result await self . hook_registry . execute ( "after_rpc_call" , rpc_message = rpc_message , result_message = result_message ) if not result_message . error : logger . info ( L ( "🏁  Remote call of {} completed in {}", Bold ( rpc_message . canonical_name ) , human_time ( call_time ) , ) ) else : logger . warning ( L ( "⚡ Error during remote call of RPC {}. Took {}: {}", Bold ( rpc_message . canonical_name ) , human_time ( call_time ) , result_message . result , ) ) raise LightbusWorkerError ( "Error while calling {}: {}\nRemote stack trace:\n{}" . format ( rpc_message . canonical_name , result_message . result , result_message . trace ) ) validate_incoming ( self . config , self . schema , result_message ) return result_message . result