<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using Elders.Cronus.MessageProcessing;
using Elders.Multithreading.Scheduler;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;

namespace Elders.Cronus.Transport.RabbitMQ
{
    public class RabbitMqConsumer<T> : IConsumer<T>
    {
        static readonly ILogger logger = CronusLogger.CreateLogger(typeof(RabbitMqConsumer<>));

        private RabbitMqConsumerOptions options;
        private readonly BoundedContext boundedContext;
        private readonly ISubscriberCollection<T> subscriberCollection;
        private WorkPool pool;
        private readonly ISerializer serializer;
        private readonly IConnectionFactory connectionFactory;

        public RabbitMqConsumer(IOptionsMonitor<RabbitMqConsumerOptions> options, IOptionsMonitor<BoundedContext> boundedContext, ISubscriberCollection<T> subscriberCollection, ISerializer serializer, IConnectionFactory connectionFactory)
        {
            if (ReferenceEquals(null, subscriberCollection)) throw new ArgumentNullException(nameof(subscriberCollection));
            if (ReferenceEquals(null, serializer)) throw new ArgumentNullException(nameof(serializer));

            this.boundedContext = boundedContext.CurrentValue;
            this.options = options.CurrentValue;
            options.OnChange(OptionsChanged);
            this.subscriberCollection = subscriberCollection;
            this.serializer = serializer;
            this.connectionFactory = connectionFactory;
        }

        protected virtual void ConsumerStart() { }
        protected virtual void ConsumerStarted() { }

        public void Start()
        {
            if (subscriberCollection.Subscribers.Any() == false)
            {
                logger.Warn($"Consumer {boundedContext}.{typeof(T).Name} not started because there are no subscribers");
                return;
            }

            ConsumerStart();

            if (pool is null == false)
            {
                logger.Warn($"RabbitMq consumer has already been started with '{options.WorkersCount}' consuments. Returning.");
                return;
            }

            var poolName = $"cronus: {typeof(T).Name}";
            pool = new WorkPool(poolName, options.WorkersCount);
            for (int i = 0; i < options.WorkersCount; i++)
            {
                var consumer = new RabbitMqContinuousConsumer<T>(boundedContext, serializer, connectionFactory, subscriberCollection);
                pool.AddWork(consumer);
            }

            pool.StartCrawlers();

            ConsumerStarted();
        }

        public void Stop()
        {
            pool.Stop();
        }

        private void OptionsChanged(RabbitMqConsumerOptions options)
        {
            this.options = options;

            Stop();
            Start();
        }
    }
}<commit_msg>Check if the WorkPool is initialized before stopping it<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using Elders.Cronus.MessageProcessing;
using Elders.Multithreading.Scheduler;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;

namespace Elders.Cronus.Transport.RabbitMQ
{
    public class RabbitMqConsumer<T> : IConsumer<T>
    {
        static readonly ILogger logger = CronusLogger.CreateLogger(typeof(RabbitMqConsumer<>));

        private RabbitMqConsumerOptions options;
        private readonly BoundedContext boundedContext;
        private readonly ISubscriberCollection<T> subscriberCollection;
        private WorkPool pool;
        private readonly ISerializer serializer;
        private readonly IConnectionFactory connectionFactory;

        public RabbitMqConsumer(IOptionsMonitor<RabbitMqConsumerOptions> options, IOptionsMonitor<BoundedContext> boundedContext, ISubscriberCollection<T> subscriberCollection, ISerializer serializer, IConnectionFactory connectionFactory)
        {
            if (ReferenceEquals(null, subscriberCollection)) throw new ArgumentNullException(nameof(subscriberCollection));
            if (ReferenceEquals(null, serializer)) throw new ArgumentNullException(nameof(serializer));

            this.boundedContext = boundedContext.CurrentValue;
            this.options = options.CurrentValue;
            options.OnChange(OptionsChanged);
            this.subscriberCollection = subscriberCollection;
            this.serializer = serializer;
            this.connectionFactory = connectionFactory;
        }

        protected virtual void ConsumerStart() { }
        protected virtual void ConsumerStarted() { }

        public void Start()
        {
            if (subscriberCollection.Subscribers.Any() == false)
            {
                logger.Warn($"Consumer {boundedContext}.{typeof(T).Name} not started because there are no subscribers");
                return;
            }

            ConsumerStart();

            if (pool is null == false)
            {
                logger.Warn($"RabbitMq consumer has already been started with '{options.WorkersCount}' consuments. Returning.");
                return;
            }

            var poolName = $"cronus: {typeof(T).Name}";
            pool = new WorkPool(poolName, options.WorkersCount);
            for (int i = 0; i < options.WorkersCount; i++)
            {
                var consumer = new RabbitMqContinuousConsumer<T>(boundedContext, serializer, connectionFactory, subscriberCollection);
                pool.AddWork(consumer);
            }

            pool.StartCrawlers();

            ConsumerStarted();
        }

        public void Stop()
        {
            pool?.Stop();
        }

        private void OptionsChanged(RabbitMqConsumerOptions options)
        {
            this.options = options;

            Stop();
            Start();
        }
    }
}
{'fraction_non_alphanumeric': 0.0809258320662274, 'fraction_numerical': 0.00033789491468153406, 'mean_word_length': 2.5091879075281565, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '933352', 'n_tokens_mistral': 1476, 'n_tokens_neox': 1388, 'n_words': 330}