<commit_before>﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Akka.Actor;
using log4net;
using RabbitMQ.Client;
using SportingSolutions.Udapi.Sdk.Interfaces;
using SportingSolutions.Udapi.Sdk.Model.Message;

namespace SportingSolutions.Udapi.Sdk.Actors
{
    /// <summary>
    /// 
    ///     The StreamController is responsible for managing the connection
    ///     to the RabbitMQ streaming server.
    /// 
    ///     There is only ONE streaming connection, independently of how
    ///     many resources/consumers are added.
    /// 
    ///     Each consumer has its own queue, but the connection is shared
    ///     among all the consumers. If the connection goes down, all
    ///     the consumers get disconnected. 
    /// 
    ///     There is no automatic re-connection. A connection is (re)-established
    ///     when the first consumer is added.
    /// 
    ///     Once a connection is established the StreamSubscriber object
    ///     is set to read from the connection for any up coming messages.
    /// 
    ///     The StreamSubscriber then passed this object to the IDispatch
    ///     object whose task it to dispatch the messages to the correct
    ///     consumer.
    /// 
    /// </summary>
    internal class StreamControllerActor : ReceiveActor, IWithUnboundedStash
    {


        internal enum ConnectionState
        {
            DISCONNECTED = 0,
            CONNECTING = 1,
            CONNECTED = 2
        }


        public const string ActorName = "StreamControllerActor";
        public int timeoutCounter = 0;
        public const int TimeoutCounterLimit = 20;
        private static ICancelable _validateCancellation;
        private static readonly ILog _logger = LogManager.GetLogger(typeof(StreamControllerActor));
        protected IConnection _streamConnection;
        private volatile ConnectionState _state;
        private readonly ICancelable _connectionCancellation = new Cancelable(Context.System.Scheduler);


        public StreamControllerActor(IActorRef dispatcherActor)
        {
            if (dispatcherActor == null)
                throw new ArgumentNullException("dispatcher");

            Dispatcher = dispatcherActor;
            timeoutCounter = 0;
            //Start in Disconnected state
            DisconnectedState();

            AutoReconnect = UDAPI.Configuration.AutoReconnect;
            CancelValidationMessages();
            _validateCancellation = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(60000, 60000, Self, new ValidateStateMessage(), Self);

            _logger.DebugFormat("StreamController initialised, AutoReconnect={0}", AutoReconnect);
        }

        private static void CancelValidationMessages()
        {

            if (_validateCancellation == null)
                return;
            _logger.Debug("CancelValidationMessages triggered");
            _validateCancellation.Cancel();
            _validateCancellation = null;
        }

        protected override void PreRestart(Exception reason, object message)
        {
            _logger.Error(
                $"Actor restart reason exception={reason?.ToString() ?? "null"}." +
                (message != null
                    ? $" last processing messageType={message.GetType().Name}"
                    : ""));

            CancelValidationMessages();

            base.PreRestart(reason, new DisconnectedMessage { IDConnection = _streamConnection?.GetHashCode() });
        }

        /// <summary>
        /// This is the state when the connection is closed
        /// </summary>
        private void DisconnectedState()
        {
            _logger.Info("Moved to DisconnectedState");<commit_msg>UseStreamControllerMailbox added to creation log<commit_after>﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Akka.Actor;
using log4net;
using RabbitMQ.Client;
using SportingSolutions.Udapi.Sdk.Interfaces;
using SportingSolutions.Udapi.Sdk.Model.Message;

namespace SportingSolutions.Udapi.Sdk.Actors
{
    /// <summary>
    /// 
    ///     The StreamController is responsible for managing the connection
    ///     to the RabbitMQ streaming server.
    /// 
    ///     There is only ONE streaming connection, independently of how
    ///     many resources/consumers are added.
    /// 
    ///     Each consumer has its own queue, but the connection is shared
    ///     among all the consumers. If the connection goes down, all
    ///     the consumers get disconnected. 
    /// 
    ///     There is no automatic re-connection. A connection is (re)-established
    ///     when the first consumer is added.
    /// 
    ///     Once a connection is established the StreamSubscriber object
    ///     is set to read from the connection for any up coming messages.
    /// 
    ///     The StreamSubscriber then passed this object to the IDispatch
    ///     object whose task it to dispatch the messages to the correct
    ///     consumer.
    /// 
    /// </summary>
    internal class StreamControllerActor : ReceiveActor, IWithUnboundedStash
    {


        internal enum ConnectionState
        {
            DISCONNECTED = 0,
            CONNECTING = 1,
            CONNECTED = 2
        }


        public const string ActorName = "StreamControllerActor";
        public int timeoutCounter = 0;
        public const int TimeoutCounterLimit = 20;
        private static ICancelable _validateCancellation;
        private static readonly ILog _logger = LogManager.GetLogger(typeof(StreamControllerActor));
        protected IConnection _streamConnection;
        private volatile ConnectionState _state;
        private readonly ICancelable _connectionCancellation = new Cancelable(Context.System.Scheduler);


        public StreamControllerActor(IActorRef dispatcherActor)
        {
            if (dispatcherActor == null)
                throw new ArgumentNullException("dispatcher");

            Dispatcher = dispatcherActor;
            timeoutCounter = 0;
            //Start in Disconnected state
            DisconnectedState();

            AutoReconnect = UDAPI.Configuration.AutoReconnect;
            CancelValidationMessages();
            _validateCancellation = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(60000, 60000, Self, new ValidateStateMessage(), Self);

            _logger.DebugFormat("StreamController initialised, AutoReconnect={0}, UseStreamControllerMailbox={1} ", AutoReconnect, UDAPI.Configuration.UseStreamControllerMailbox);
        }

        private static void CancelValidationMessages()
        {

            if (_validateCancellation == null)
                return;
            _logger.Debug("CancelValidationMessages triggered");
            _validateCancellation.Cancel();
            _validateCancellation = null;
        }

        protected override void PreRestart(Exception reason, object message)
        {
            _logger.Error(
                $"Actor restart reason exception={reason?.ToString() ?? "null"}." +
                (message != null
                    ? $" last processing messageType={message.GetType().Name}"
                    : ""));

            CancelValidationMessages();

            base.PreRestart(reason, new DisconnectedMessage { IDConnection = _streamConnection?.GetHashCode() });
        }

        /// <summary>
        /// This is the state when the connection is closed
        /// </summary>
        private void DisconnectedState()
        {
            _logger.Info("Moved to DisconnectedState");
{'fraction_non_alphanumeric': 0.08205602536997886, 'fraction_numerical': 0.005153276955602537, 'mean_word_length': 2.201776649746193, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '5561129', 'n_tokens_mistral': 1872, 'n_tokens_neox': 1765, 'n_words': 533}