Este paquete lo desarrolle para un proyecto de un cliente que requería realizar diversas tareas asíncronas , pero también aquellas que requería respuesta por parte del consumidor de la cola, para ello explotamos la propiedad RPC que dispone RabbitMQ.
La lógica es sencilla el cliente realiza una petición y se adjunta a la cabecera una propiedad denominada correlation_id, el cliente espera a recibir de vuelta el mensaje con el correlation_id enviado también adjuntado a la cabecera del mensaje.
Por ejemplo imaginemos que queremos recibir el número de factura desde otro servicio lo podríamos realizar de la siguiente manera:
$this->result = $this->rabbitMQ->requestRpc(
$id,
"billing",
"billing_return",
"command.billing",
"billing.1.query.income.getNextIncomeInvoiceNumber",
json_encode($json)
);
El cliente que debe emitir la respuesta, quedaría de la siguiente manera,.
public function handle(IncomeGotNextInvoiceNumberEvent $incomeGotNextInvoiceNumberEvent)
{
$incomeNextInvoiceNumber = json_encode($incomeGotNextInvoiceNumberEvent->incomeNextInvoiceNumber());
$msg = $incomeGotNextInvoiceNumberEvent->message();
$this->rabbitMQ->responseRpc($incomeGotNextInvoiceNumberEvent->incomeNextInvoiceNumber(), $msg);
}
El correlation_id va encapsulado dentro del $msg, por tanto el cliente que solicita se queda a la espera de recibir un mensaje con ese correlation_id, en la cola y el exchange correspondiente.
Podeís encontrar el paquete en el siguiente enlace de github El proyecto también está subido en packagist para su uso con más de 1000 descargas. Y aquí os dejo un vídeo de explicación de como funciona internamente. Y en este otro repo de github el código utilizado en el ejemplo del vídeo. Lo cierto es que lo tengo abandonado pero quiero retomarlo y os animo a que me ayudéis a mejorarlo y ampliar funcionalidades, con las últimas mejoras incluidas por RabbitMQ, como puede ser los streams.
Nos vemos en un próximo post.