using System.Threading;
using System.Threading.Tasks;
using Nimbus.ConcurrentCollections;
using Nimbus.Configuration.Settings;
using Nimbus.Infrastructure;
using Nimbus.Infrastructure.MessageSendersAndReceivers;
using Nimbus.Transports.InProcess.QueueManagement;

namespace Nimbus.Transports.InProcess.MessageSendersAndReceivers
{
    internal class InProcessQueueReceiver : ThrottlingMessageReceiver
    {
        private readonly string _queuePath;
        private readonly InProcessMessageStore _messageStore;

        private readonly ThreadSafeLazy<AsyncBlockingCollection<NimbusMessage>> _messageQueue;

        public InProcessQueueReceiver(string queuePath,
                                      ConcurrentHandlerLimitSetting concurrentHandlerLimit,
                                      InProcessMessageStore messageStore,
                                      IGlobalHandlerThrottle globalHandlerThrottle,
                                      ILogger logger) : base(concurrentHandlerLimit, globalHandlerThrottle, logger)
        {
            _queuePath = queuePath;
            _messageStore = messageStore;

            _messageQueue = new ThreadSafeLazy<AsyncBlockingCollection<NimbusMessage>>(() => _messageStore.GetOrCreateMessageQueue(_queuePath));
        }

        public override string ToString()
        {
            return _queuePath;
        }

        protected override Task WarmUp()
        {
            _messageQueue.EnsureValueCreated();

            return base.WarmUp();
        }

        protected override async Task<NimbusMessage> Fetch(CancellationToken cancellationToken)
        {
            var nimbusMessage = await _messageQueue.Value.Take(cancellationToken);
            return nimbusMessage;
        }
    }
}
using System.Threading;
using System.Threading.Tasks;
using Nimbus.ConcurrentCollections;
using Nimbus.Configuration.Settings;
using Nimbus.Infrastructure;
using Nimbus.Infrastructure.MessageSendersAndReceivers;
using Nimbus.Transports.InProcess.QueueManagement;

namespace Nimbus.Transports.InProcess.MessageSendersAndReceivers
{
    internal class InProcessQueueReceiver : ThrottlingMessageReceiver
    {
        private readonly string _queuePath;
        private readonly InProcessMessageStore _messageStore;

        private readonly ThreadSafeLazy<AsyncBlockingCollection<NimbusMessage>> _messageQueue;

        public InProcessQueueReceiver(string queuePath,
                                      ConcurrentHandlerLimitSetting concurrentHandlerLimit,
                                      InProcessMessageStore messageStore,
                                      IGlobalHandlerThrottle globalHandlerThrottle,
                                      ILogger logger) : base(concurrentHandlerLimit, globalHandlerThrottle, logger)
        {
            _queuePath = queuePath;
            _messageStore = messageStore;

            _messageQueue = new ThreadSafeLazy<AsyncBlockingCollection<NimbusMessage>>(() => _messageStore.GetOrCreateMessageQueue(_queuePath));
        }

        public override string ToString()
        {
            return _queuePath;
        }

        protected override Task WarmUp()
        {
            return Task.Run(() => _messageQueue.EnsureValueCreated());
        }

        protected override async Task<NimbusMessage> Fetch(CancellationToken cancellationToken)
        {
            var nimbusMessage = await _messageQueue.Value.Take(cancellationToken);
            return nimbusMessage;
        }
    }
}
Revert "Included call to base.WarmUp()."
Revert "Included call to base.WarmUp()."

This reverts commit c33d7b0c767eea6dc7b341865b2fa0cd800962c6.
