RabbitMQ RPC Direct Reply-To in C#

README.md

https://www.rabbitmq.com/tutorials/tutorial-six-dotnet.html https://www.rabbitmq.com/direct-reply-to.html https://www.cloudamqp.com/blog/2015-09-03-part4-rabbitmq-for-beginners-exchanges-routing-keys-bindings.html

Client.cs

using System;
using System.Collections.Concurrent;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Client
{
    public class RpcClient
    {
        private readonly IConnection connection;
        private readonly IModel channel;
        private readonly string queueName;
        private readonly EventingBasicConsumer consumer;
        private readonly BlockingCollection<string> respQueue = new BlockingCollection<string>();
        private readonly IBasicProperties props;

        public RpcClient(string queueName)
        {
            this.queueName = queueName;

            var factory = new ConnectionFactory()
            {
                HostName = "TODO",
                UserName = "TODO",
                Password = "TODO",
            };

            connection = factory.CreateConnection();
            channel = connection.CreateModel();
            consumer = new EventingBasicConsumer(channel);

            props = channel.CreateBasicProperties();
            var correlationId = Guid.NewGuid().ToString();
            props.CorrelationId = correlationId;
            props.ReplyTo = "amq.rabbitmq.reply-to";

            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var response = Encoding.UTF8.GetString(body);
                if (ea.BasicProperties.CorrelationId == correlationId)
                {
                    respQueue.Add(response);
                }
            };
        }

        public string Call(string message)
        {
            var messageBytes = Encoding.UTF8.GetBytes(message);

            // The RPC client must consume in the automatic acknowledgement mode.
            channel.BasicConsume(
                consumer: consumer,
                queue: "amq.rabbitmq.reply-to",
                autoAck: true);

            channel.BasicPublish(
                exchange: "",
                routingKey: this.queueName,
                basicProperties: props,
                body: messageBytes);


            return respQueue.Take();
        }

        public void Close()
        {
            connection.Close();
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            var queue = $"HealthCheck_W{args[0]}";
            
            do
            {
                var rpcClient = new RpcClient(queue);
                Console.WriteLine(" [x] Requesting fib(30)");
                var response = rpcClient.Call("30");

                Console.WriteLine(" [.] Got '{0}'", response);
                rpcClient.Close();

                Console.WriteLine("Do it again? Y/n");
            } while (Char.ToUpper(Console.ReadKey().KeyChar) == 'Y');

        }
    }
}

run-client.cmd

dotnet run -p .\Client

run-server.cmd

dotnet run -p .\Server -- 1

Server.cs

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Server
{
    class Program
    {
        public static void Main(string[] args)
        {
            var queue = $"HealthCheck_W{args[0]}";
            
            var factory = new ConnectionFactory()
            {
                HostName = "TODO",
                UserName = "TODO",
                Password = "TODO",
            };

            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(
                    queue: queue,
                    durable: false,
                    exclusive: false,
                    autoDelete: false,
                    arguments: null);
                channel.BasicQos(0, 1, false);
                
                // Configure request consumer
                var consumer = new EventingBasicConsumer(channel);
                
                // ???? Consume from the pseudo-queue amq.rabbitmq.reply-to in no-ack mode.
                channel.BasicConsume(queue: queue, autoAck: false, consumer: consumer);
                consumer.Received += (model, ea) =>
                {
                    string response = null;

                    var body = ea.Body;
                    var props = ea.BasicProperties;
                    var replyProps = channel.CreateBasicProperties();
                    replyProps.CorrelationId = props.CorrelationId;

                    try
                    {
                        var message = Encoding.UTF8.GetString(body);
                        int n = int.Parse(message);
                        Console.WriteLine(" [.] fib({0})", message);
                        response = fib(n).ToString();
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine(" [.] " + e.Message);
                        response = "";
                    }
                    finally
                    {
                        var responseBytes = Encoding.UTF8.GetBytes(response);
                        channel.BasicPublish(
                            exchange: "",                // Must reply to default exchange ("")
                            routingKey: props.ReplyTo,
                            basicProperties: replyProps,
                            body: responseBytes);
                        channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                    }
                };

                Console.WriteLine(" [x] Awaiting RPC requests");
                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }
        }

        private static int fib(int n)
        {
            return 832040;
            
            if (n == 0 || n == 1)
            {
                return n;
            }

            return fib(n - 1) + fib(n - 2);
        }
    }
}

Is it ok to use the same channel for BasicConsume, BasicPublish and BasicAck? In the documentation they say it's not thread safe. If I understand it correctly even with synchronous handler, which is inefficient on top of that.