6
\$\begingroup\$

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();
        }
    }
}
\$\endgroup\$
9
\$\begingroup\$

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 ?.

\$\endgroup\$
  • \$\begingroup\$ 2. I actually thought about using Lazy<T> but the reason that I did not was that I need to use some runtime variables to create the connection and not just static variable (It didn't show in the code that I wrote here - sorry about that) \$\endgroup\$ – Brian Pilgren Mortensen Jan 10 '17 at 12:25
  • \$\begingroup\$ 3. Very good point! I have updated my code and I have extracted the connection into a separate class that is injected into both the Consumer and Producer. It will then be up the application to handle the lifetime of the connection. This also makes the Dispose a lot more simple and I can avoid the points that you mentioned in 1 \$\endgroup\$ – Brian Pilgren Mortensen Jan 10 '17 at 12:26
  • \$\begingroup\$ 4. The thread safty I am looking for is mostly about the connection and handling the creation/disposing of this while used by several threads. I think that this is much more clear now with the solution you mentioned in 3. I know that the IModel is not thread safe and therefor the RmqMessageQueueConsumer is not thread safe either - I will asume that only one thread is accessing each instance. \$\endgroup\$ – Brian Pilgren Mortensen Jan 10 '17 at 12:26
  • \$\begingroup\$ @BrianPilgrenMortensen Also. What happens if the applications crashes - on Windows when application crashes pretty much everything is released. That includes memory, handles, socket connections, mutexes, etc. However, any un-saved internal state is lost. Your services won't magically get finalized, garbage collected, closed, saved or disposed. You will have to do this manually in exception handler (if you need to). \$\endgroup\$ – Nikita B Jan 10 '17 at 13:18

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.