I just implemented an application using RabbitMQ RPC. I need to call multiple methods and consume the results of them. So, created generic classes and methods to reach my goal.
The main problem is: My client side consumer, consumes every response more than one time, because this consumer listen to just one queue ('amq.rabbitmq.reply-to' or any named queue), but I have to add consumer for my every model.
I could handle this problem using CorrelationId property of messages, but the problem will become more serious when I have lots of models to consume. At first I followed the official sample of RabbitMQ, then implemented this approach, and now I'm using Direct Reply-to approach.
This is how I implement my consumer:
RpcClient.consumer.Received += (model, ea) =>
{
if (!_callbackMapper.TryRemove(ea.BasicProperties.CorrelationId, out TaskCompletionSource<TResponse> tcs))
return;
var body = ea.Body.ToArray();
var response = body.FromByteArray<TResponse>();
tcs.TrySetResult(response);
};
This code snippet runs just one time for every model. Note: This code snippet must not run just one time, because it will listen just to one model.