Asynchronous processing

JsonRpcBundle accepts RPC requests not only over HTTP. The same methods can be called through Symfony Messenger message queues, and even in synchronous requests the server performs part of the work outside the main flow.

Asynchronous processing in synchronous requests

• Batch requests in process mode execute items in parallel in child processes.
• Notifications and requests with $rpc.callback are executed after the response has been sent to the client.

Asynchronous transport (Symfony Messenger)

Transports are declared as a list in the async block of the configuration:

# config/packages/ufo_json_rpc.yaml
ufo_json_rpc:
    async:
        - type: 'amqp'
          config:
              name: 'rpc_async'
              dsn: '%env(resolve:RPC_TRANSPORT_DSN)%'

Items with type: amqp are registered as Symfony Messenger transports: name becomes the transport name, and RpcAsyncRequest messages are routed to it (or to config.route, if set). The DSN can point to any transport supported by Messenger.

To run the RPC server as a queue consumer, run:

php bin/console messenger:consume rpc_async

The consumer validates the token and executes the method. Execution events are dispatched (rpc.pre_execute, rpc.pre_response, rpc.post_execute, rpc.error), but not rpc.request and rpc.post_response — see Lifecycle. Token errors and unknown method errors become Messenger message handling errors (retries, failure transport).

Sending a request to the queue

use Symfony\Component\Messenger\MessageBusInterface;
use Ufo\RpcObject\RpcAsyncRequest;

$bus->dispatch(new RpcAsyncRequest($rpcRequest, $token, apiGroup: 'internal'));

apiGroup defines the API group; without it, general is used. The token is validated by the same validator as for HTTP.

Other transport types

Items with another type (for example, mercure) are not connected to Messenger by the bundle. They are available to the application via RpcAsyncConfig::getConfig(<type>) and, if docs.async_dsn_info: true, are published in the documentation (x-ufo.transport) under their name.

ufo_json_rpc:
    async:
        - type: 'mercure'
          config:
              name: 'rpc_socket'
              dsn: '%env(resolve:MERCURE_PUBLIC_URL)%'
              topics_prefix: 'rpc.event.'

Advantages of asynchronous transport

1. Scalability: the load is distributed among several consumers.
2. Reliability: the broker (RabbitMQ, etc.) guarantees message delivery.
3. Single API: the same methods, validation and security as for HTTP.