<commit_before>using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.ServiceBus.Messaging;
using Nimbus.Configuration.Settings;

namespace Nimbus.Infrastructure.MessageSendersAndReceivers
{
    internal abstract class NimbusMessageReceiver : INimbusMessageReceiver
    {
        protected readonly ConcurrentHandlerLimitSetting ConcurrentHandlerLimit;
        private readonly ILogger _logger;
        private bool _running;

        private readonly object _mutex = new object();
        private Task _workerTask;
        private readonly SemaphoreSlim _throttle;
        private readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();

        protected NimbusMessageReceiver(ConcurrentHandlerLimitSetting concurrentHandlerLimit, ILogger logger)
        {
            ConcurrentHandlerLimit = concurrentHandlerLimit;
            _logger = logger;
            _throttle = new SemaphoreSlim(concurrentHandlerLimit, concurrentHandlerLimit);
        }

        public async Task Start(Func<BrokeredMessage, Task> callback)
        {
            if (_running) throw new InvalidOperationException("Already started!");
            _running = true;

            await CreateBatchReceiver();
            _workerTask = Task.Run(() => Worker(callback));
        }

        protected abstract Task CreateBatchReceiver();
        protected abstract Task<BrokeredMessage[]> FetchBatch(int batchSize);
        protected abstract void StopBatchReceiver();

        private async Task Worker(Func<BrokeredMessage, Task> callback)
        {
            while (_running)
            {
                try
                {
                    var messages = await FetchBatchInternal(_throttle.CurrentCount);
                    if (!_running) return;

                    var tasks = messages
                        .Select(async m =>
                                      {
                                          try
                                          {
                                              await _throttle.WaitAsync(_cancellationToken.Token);
                                              await callback(m);
                                          }
                                          finally
                                          {
                                              _throttle.Release();
                                          }
                                      })
                        .ToArray();

                    await Task.WhenAny(tasks);
                }<commit_msg>Fixed dispatcher bug when there are zero messages returned from a BatchReceive call. We were trying to await WhenAny() the dispatcher task collection but there were no tasks because there were no messages.<commit_after>using System;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.ServiceBus.Messaging;
using Nimbus.Configuration.Settings;
using Nimbus.Extensions;

namespace Nimbus.Infrastructure.MessageSendersAndReceivers
{
    internal abstract class NimbusMessageReceiver : INimbusMessageReceiver
    {
        protected readonly ConcurrentHandlerLimitSetting ConcurrentHandlerLimit;
        private readonly ILogger _logger;
        private bool _running;

        private readonly object _mutex = new object();
        private Task _workerTask;
        private readonly SemaphoreSlim _throttle;
        private readonly CancellationTokenSource _cancellationToken = new CancellationTokenSource();

        protected NimbusMessageReceiver(ConcurrentHandlerLimitSetting concurrentHandlerLimit, ILogger logger)
        {
            ConcurrentHandlerLimit = concurrentHandlerLimit;
            _logger = logger;
            _throttle = new SemaphoreSlim(concurrentHandlerLimit, concurrentHandlerLimit);
        }

        public async Task Start(Func<BrokeredMessage, Task> callback)
        {
            if (_running) throw new InvalidOperationException("Already started!");
            _running = true;

            await CreateBatchReceiver();
            _workerTask = Task.Run(() => Worker(callback));
        }

        protected abstract Task CreateBatchReceiver();
        protected abstract Task<BrokeredMessage[]> FetchBatch(int batchSize);
        protected abstract void StopBatchReceiver();

        private async Task Worker(Func<BrokeredMessage, Task> callback)
        {
            while (_running)
            {
                try
                {
                    var messages = await FetchBatchInternal(_throttle.CurrentCount);
                    if (!_running) return;
                    if (messages.None()) continue;

                    var tasks = messages
                        .Select(async m =>
                                      {
                                          try
                                          {
                                              await _throttle.WaitAsync(_cancellationToken.Token);
                                              await callback(m);
                                          }
                                          finally
                                          {
                                              _throttle.Release();
                                          }
                                      })
                        .ToArray();

                    await Task.WhenAny(tasks);
                }
{'fraction_non_alphanumeric': 0.05535190615835777, 'fraction_numerical': 0.0, 'mean_word_length': 1.4536870503597121, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3875501', 'n_tokens_mistral': 1086, 'n_tokens_neox': 1064, 'n_words': 287}