<commit_before>﻿using MessageQueuer.Core;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;

namespace MessageQueuer
{
    public class MqRunner
    {
        private static readonly object Lock = new object();

        private readonly MqConfiguration _configuration;
        private readonly MqRecieverInvoker _recieverInvoker;
        private readonly TypeLocator _typeLocator;

        private readonly ConcurrentBag<MqQueue> _queues = new ConcurrentBag<MqQueue>();
        private readonly ConcurrentBag<MqHandler> _handlers = new ConcurrentBag<MqHandler>();

        private bool _isInitialized;

        public MqRunner(MqConfiguration configuration)
        {
            _configuration = configuration;
            _typeLocator = new TypeLocator();
            _recieverInvoker = new MqRecieverInvoker(new InvokerFactory(configuration.Resolver), configuration.Serializer);
        }

        public void Start()
        {
            Start(null);
        }

        public void Start(Action<Exception> onException)
        {
            if (!_isInitialized)
                Initialize();

            Task.Factory.StartNew(() =>
            {
                Parallel.ForEach(_queues, (queue, state) =>
                {
                    var handler = new MqHandler(_configuration, queue, _recieverInvoker);
                    handler.Start();

                    _handlers.Add(handler);
                });
            }, TaskCreationOptions.LongRunning)
            .ContinueWith(x =>
            {
                lock (Lock)
                {
                    if (!x.IsFaulted) return;

                    // Lets stop the runner
                    Stop();

                    // Invoke the onException method
                    if (x.Exception != null)
                    {
                        var flatException = x.Exception.Flatten();

                        if (flatException.InnerException != null && flatException.InnerException.InnerException != null)
                            onException.Invoke(flatException.InnerException.InnerException);
                        else if (flatException.InnerException != null)
                            onException.Invoke(flatException.InnerException);
                        else
                            onException.Invoke(flatException);
                    }
                }
            }, TaskContinuationOptions.OnlyOnFaulted);
        }

        public void Stop()
        {
            while (!_handlers.IsEmpty)
            {
                MqHandler handler;

                if (!_handlers.TryTake(out handler))
                {
                    break;
                }

                if (handler.IsRunning)<commit_msg>The running failed when 'onException' null '<commit_after>﻿using MessageQueuer.Core;
using System;
using System.Collections.Concurrent;
using System.Linq;
using System.Threading.Tasks;

namespace MessageQueuer
{
    public class MqRunner
    {
        private static readonly object Lock = new object();

        private readonly MqConfiguration _configuration;
        private readonly MqRecieverInvoker _recieverInvoker;
        private readonly TypeLocator _typeLocator;

        private readonly ConcurrentBag<MqQueue> _queues = new ConcurrentBag<MqQueue>();
        private readonly ConcurrentBag<MqHandler> _handlers = new ConcurrentBag<MqHandler>();

        private bool _isInitialized;

        public MqRunner(MqConfiguration configuration)
        {
            _configuration = configuration;
            _typeLocator = new TypeLocator();
            _recieverInvoker = new MqRecieverInvoker(new InvokerFactory(configuration.Resolver), configuration.Serializer);
        }

        public void Start()
        {
            Start(null);
        }

        public void Start(Action<Exception> onException)
        {
            if (!_isInitialized)
                Initialize();

            Task.Factory.StartNew(() =>
            {
                Parallel.ForEach(_queues, (queue, state) =>
                {
                    var handler = new MqHandler(_configuration, queue, _recieverInvoker);
                    handler.Start();

                    _handlers.Add(handler);
                });
            }, TaskCreationOptions.LongRunning)
            .ContinueWith(x =>
            {
                lock (Lock)
                {
                    if (!x.IsFaulted) return;

                    // Lets stop the runner
                    Stop();

                    // Invoke the onException method
                    if (onException != null && x.Exception != null)
                    {
                        var flatException = x.Exception.Flatten();

                        if (flatException.InnerException != null && flatException.InnerException.InnerException != null)
                            onException.Invoke(flatException.InnerException.InnerException);
                        else if (flatException.InnerException != null)
                            onException.Invoke(flatException.InnerException);
                        else
                            onException.Invoke(flatException);
                    }
                }
            }, TaskContinuationOptions.OnlyOnFaulted);
        }

        public void Stop()
        {
            while (!_handlers.IsEmpty)
            {
                MqHandler handler;

                if (!_handlers.TryTake(out handler))
                {
                    break;
                }

                if (handler.IsRunning)
{'fraction_non_alphanumeric': 0.07553571428571429, 'fraction_numerical': 0.0, 'mean_word_length': 1.4772224679345423, 'pattern_counts': {'":': 0, '<': 13, '<?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': '5474632', 'n_tokens_mistral': 1326, 'n_tokens_neox': 1250, 'n_words': 275}