RabbitMQ wrapper to work in a concurrent environment

Question

I'm developing a common library for my organisation that will be used across different services by different teams. The library is a simple wrapper for RabbitMQ.

Some services will only be producers, some will only be consumers and some will be both. For that reason I have created a Consumer and a Producer. Each of these will be implementations of an abstract base class.

The Producer should be able to work in a multi threaded environment, e.g. ASP.NET MVC, either as a singleton or separate instance for each thread.

It should be possible to start several instances of the Consumer, either on separate threads or on the same thread.

In RabbitMQ each application should only create one connection and then reuse this connection for all communication. As the Consumer and Procucer cannot know if the connection can be closed it has to be done in the Dispose method.

I want to make the implementations as safe as possible to make sure that the developers using the library cannot accidentally create resource leaks or concurrence issues.

The implementation works but I'm not sure if there are any concurrency pitfalls that I have not thought about that could make this blow up.

Can this implementation handle the connection being disposed and recreated in a concurrent environment? What happens if the applications crashes and nothing handles this, will the Garbage Collector clean up the connection/channels?

I'm not all that familiar with this type of programming so I would like to know if this implementations is OK or if I have missed something?

The two interfaces for the Consumer and Producer

public interface IMessageQueueConsumer
{
    void StartConsuming(IList<IMessageListener> messageListeners);
}

public interface IMessageQueueProducer
{
    void Publish<T>(T message) where T : OutgoingMessage;
}

The base class

public abstract class RmqBase : IDisposable
{
    private static volatile IConnection _connection;
    private static readonly object ConnectionLock = new object();

    protected IConnection Connection
    {
        get
        {
            if (_connection != null)
            {
                return _connection;
            }
            lock (ConnectionLock)
            {
                if (_connection != null)
                {
                    return _connection;
                }
                var connectionFactory = new ConnectionFactory
                {
                    UserName = "testUser",
                    Password = "testPass",
                    VirtualHost = "testHost",
                    HostName = "testHostName",
                    Port = 1324,
                    AutomaticRecoveryEnabled = true
                };
                _connection = connectionFactory.CreateConnection();
            }
            return _connection;
        }
    }

    public void Dispose()
    {
        Dispose(true);
        GC.SuppressFinalize(this);
    }

    protected virtual void Dispose(bool disposing)
    {
        if (!disposing)
        {
            return;
        }
        if (_connection == null)
        {
            return;
        }
        lock (ConnectionLock)
        {
            if (_connection == null)
            {
                return;
            }
            _connection?.Dispose();
            _connection = null;
        }
    }
}

The Consumer

public sealed class RmqMessageQueueConsumer : RmqBase, IMessageQueueConsumer
{
    public void StartConsuming(IList<IMessageListener> messageListeners)
    {
        ConsumeChannel.ExchangeDeclare("myExchange", RabbitMQ.Client.ExchangeType.Direct, true, false, null);
        ConsumeChannel.QueueDeclare("myQueue", true, false, false, null);
        foreach (var messageListener in messageListeners)
        {
            ConsumeChannel.QueueBind("myQueue", "myExchange", messageListener.MessageName.Trim().ToLower());
        }
        var consumer = new EventingBasicConsumer(ConsumeChannel);
        consumer.Received += (model, eventArgs) =>
        {
            MessageHandler.HandleMessage(eventArgs);
            ConsumeChannel.BasicAck(eventArgs.DeliveryTag, false);
        };
    }

    private volatile IModel _consumeChannel;
    private readonly object _connectionLock = new object();

    private IModel ConsumeChannel
    {
        get
        {
            if (_consumeChannel != null)
            {
                return _consumeChannel;
            }
            lock (_connectionLock)
            {
                if (_consumeChannel != null)
                {
                    return _consumeChannel;
                }
                _consumeChannel = Connection.CreateModel();
            }
            return _consumeChannel;
        }
    }

    protected override void Dispose(bool disposing)
    {
        if (disposing && _consumeChannel != null)
        {
            lock (_connectionLock)
            {
                if (_consumeChannel != null)
                {
                    if (!_consumeChannel.IsClosed)
                    {
                        _consumeChannel.Close(200, "StopConsuming");
                    }
                    _consumeChannel?.Dispose();
                    _consumeChannel = null;
                }
            }
        }
        base.Dispose(disposing);
    }
}

The Producer

public sealed class RmqMessageQueueProducer : RmqBase, IMessageQueueProducer
{
    public void Publish<T>(T message) where T : OutgoingMessage
    {
        var properties = new BasicProperties
        {
            Persistent = true,
            MessageId = message.Id,
            Timestamp = new AmqpTimestamp(message.TimeStamp.ToUnixTimeMilliseconds()),
            CorrelationId = message.CorrelationId,
            ContentType = "application/json",
            ContentEncoding = Encoding.UTF8.WebName
        };
        var body = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message.Data));

        using (var publishChannel = Connection.CreateModel())
        {
            publishChannel.ExchangeDeclare("myExchange", RabbitMQ.Client.ExchangeType.Direct, true, false, null);
            publishChannel.ConfirmSelect();
            publishChannel.BasicPublish("myExchange", message.Name, properties, body);
            publishChannel.WaitForConfirmsOrDie();
        }
    }
}

Answer 1

A few things that caught my eye:

1. So called dispose pattern only makes sense if you implement a finalizer on your base class or on one of its descendants. If you do not finalize anything - don't bother with passing bool flag, that is always true. YAGNI.
2. Always use the highest level tool available to you, especially if you are not comfortable with multi-threading. Instead of manually creating models and connections - use Lazy<T> instead.
3. Disposing static instance in non-static method is extremely unsafe. Disposing single service will close connection for all other services, even if the connection is currently in use. I think you should do one of two things. Either match connection lifetime with application lifetime: create single connection at composition level, inject it into every service, dispose it when application shuts down, after all other services are disposed. Or implement reference counting, so that connection is only disposed when no service is holding a "reference" to it.
4. "Thread safety" makes little sense, unless you specify context. For example, a quick google search told me, that IModel was not thread safe. Which, I assume, means that RmqMessageQueueConsumer.Start method won't be thread safe in a sense, that you will not be able to safely call it in parallel from different threads on the same instance. Is that the kind of thread-safety you are after?

5. You perform null check twice:

       if (_connection == null)
       {
           return;
       }
       _connection?.Dispose();
       _connection = null;
   

   either remove the first check or use . instead of ?.