<commit_before>﻿using Castle.Core.Logging;
using Jarvis.Framework.Kernel.Support;
using Metrics;
using NEventStore;
using NEventStore.Client;
using NEventStore.Persistence;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace Jarvis.Framework.Kernel.ProjectionEngine.Client
{
    public interface IPollingClientConsumer
    {
        void Consume(ICommit commit);
    }

    public class CommitPollingClient
    {
        private const string command_poll = "poll";
        private const string command_stop = "stop";
        private int _bufferSize = 4000;
        private readonly int _interval;
        readonly ICommitEnhancer _enhancer;
        private String _id;

        readonly ILogger _logger;
        private IPersistStreams _persistStreams;

        public CommitPollingClient(
            IPersistStreams persistStreams,
            ICommitEnhancer enhancer,
            String id,
            ILogger logger)
        {
            if (persistStreams == null)
            {
                throw new ArgumentNullException("persistStreams");
            }
            _persistStreams = persistStreams;
            _enhancer = enhancer;
            _consumers = new List<ITargetBlock<ICommit>>();
            _commandList = new BlockingCollection<string>();
            _logger = logger;
            _id = id;
        }

        public void AddConsumer(IPollingClientConsumer consumer)
        {
            AddConsumer(commit => consumer.Consume(commit));
        }

        public void AddConsumer(Action<ICommit> consumerAction)
        {
            ExecutionDataflowBlockOptions consumerOptions = new ExecutionDataflowBlockOptions();
            consumerOptions.BoundedCapacity = _bufferSize;
            Action<ICommit> wrapperAction = commit => 
            {
                try
                {
                    consumerAction(commit);
                }
                catch (Exception ex)
                {
                    _logger.ErrorFormat(ex, "CommitPollingClient {0}: Error during Commit consumer {1} ", _id, ex.Message);
                    CloseEverything();
                    throw;
                }
            };
            var actionBlock = new ActionBlock<ICommit>(wrapperAction, consumerOptions);
            
            _consumers.Add(actionBlock);
        }

        #region Dispatching

        private List<ITargetBlock<ICommit>> _consumers;

        #endregion

        #region polling

        private BlockingCollection<String> _commandList;

        private Thread _pollerThread;

        private System.Timers.Timer _pollerTimer;

        private String _checkpointTokenCurrent;

        private CancellationTokenSource _stopRequested = new CancellationTokenSource();

        private Int64 _lastActivityTickCount;

        public void StartAutomaticPolling(
            String checkpointTokenFrom,
            Int32 intervalInMilliseconds,
            Int32 bufferSize = 4000,
            String pollerName = "CommitPollingClient")
        {
            StartManualPolling(checkpointTokenFrom, bufferSize, pollerName);
            _pollerTimer = new System.Timers.Timer(intervalInMilliseconds);
            _pollerTimer.Elapsed += TimerCallback;
            _pollerTimer.Start();

            //Add an immediate poll
            _commandList.Add(command_poll);

            //Set health check for polling
            HealthChecks.RegisterHealthCheck("Polling-" + pollerName, () =>
            {
                if (_pollerTimer == null)
                {
                    //poller is stopped, system healty
                    return HealthCheckResult.Healthy("Automatic polling stopped");
                }
                var elapsed = Math.Abs(Environment.TickCount - _lastActivityTickCount);
                if (elapsed > 5000)
                {
                    //more than 5 seconds without a poll, polling probably is stopped
                    return HealthCheckResult.Unhealthy(String.Format("poller stuck, last polling {0} ms ago", elapsed));
                }

                return HealthCheckResult.Healthy("Poller alive");
            });
        }

        public void StartManualPolling(
            String checkpointTokenFrom,
            Int32 bufferSize = 4000,
            String pollerName = "CommitPollingClient")
        {
            _bufferSize = bufferSize;
            _checkpointTokenCurrent = checkpointTokenFrom;
            //prepare single poller thread.
            CreateTplChain();
            _pollerThread = new Thread(PollerFunc);
            _pollerThread.IsBackground = true;
            _pollerThread.Name = pollerName;
            _pollerThread.Start();
            MetricsHelper.SetCommitPollingClientBufferSize(pollerName, () => GetClientBufferSize());
        }

        public void StopPolling()
        {
            if (_pollerTimer != null)
            {
                _pollerTimer.Stop();
                _pollerTimer.Dispose();
                _pollerTimer = null;
            }
            CloseEverything();
        }

        private void CloseEverything()
        {
            _stopRequested.Cancel();
            CloseTplChain();
        }

        private void TimerCallback(object sender, System.Timers.ElapsedEventArgs e)
        {
            Poll();
        }

        /// <summary>
        /// Add a manual poll command, it will be executed from the single thread that
        /// read commits from stream.
        /// </summary>
        public void Poll()
        {
            //Add poll command if there are no other poll command in queuel
            if (!_commandList.Any(c => c.Equals(command_poll, StringComparison.OrdinalIgnoreCase)))
                _commandList.Add(command_poll);
        }

        private double GetClientBufferSize()
        {
            if (_buffer != null)
                return _buffer.Count;
            return 0;
        }
<commit_msg>Fixed healt check and poller stop on error on consumer.<commit_after>﻿using Castle.Core.Logging;
using Jarvis.Framework.Kernel.Support;
using Metrics;
using NEventStore;
using NEventStore.Client;
using NEventStore.Persistence;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace Jarvis.Framework.Kernel.ProjectionEngine.Client
{
    public interface IPollingClientConsumer
    {
        void Consume(ICommit commit);
    }


    public enum CommitPollingClientStatus
    {
        Stopped,
        Polling,
        Faulted
    }

    public class CommitPollingClient
    {
        private const string command_poll = "poll";
        private const string command_stop = "stop";
        private int _bufferSize = 4000;
        private readonly int _interval;
        readonly ICommitEnhancer _enhancer;
        private String _id;

        readonly ILogger _logger;
        private IPersistStreams _persistStreams;

        public CommitPollingClientStatus Status { get; private set; }

        public Exception LastException { get; private set; }

        public CommitPollingClient(
            IPersistStreams persistStreams,
            ICommitEnhancer enhancer,
            String id,
            ILogger logger)
        {
            if (persistStreams == null)
            {
                throw new ArgumentNullException("persistStreams");
            }
            _persistStreams = persistStreams;
            _enhancer = enhancer;
            _consumers = new List<ITargetBlock<ICommit>>();
            _commandList = new BlockingCollection<string>();
            _logger = logger;
            _id = id;
            Status = CommitPollingClientStatus.Stopped;
        }

        public void AddConsumer(IPollingClientConsumer consumer)
        {
            AddConsumer(commit => consumer.Consume(commit));
        }

        public void AddConsumer(Action<ICommit> consumerAction)
        {
            ExecutionDataflowBlockOptions consumerOptions = new ExecutionDataflowBlockOptions();
            consumerOptions.BoundedCapacity = _bufferSize;
            Action<ICommit> wrapperAction = commit => 
            {
                try
                {
                    consumerAction(commit);
                }
                catch (Exception ex)
                {
                    _logger.ErrorFormat(ex, "CommitPollingClient {0}: Error during Commit consumer {1} ", _id, ex.Message);
                    StopPolling(true); //Stop and close everything on this poller.
                    LastException = ex;
                    throw;
                }
            };
            var actionBlock = new ActionBlock<ICommit>(wrapperAction, consumerOptions);
            
            _consumers.Add(actionBlock);
        }

        #region Dispatching

        private List<ITargetBlock<ICommit>> _consumers;

        #endregion

        #region polling

        private BlockingCollection<String> _commandList;

        private Thread _pollerThread;

        private System.Timers.Timer _pollerTimer;

        private String _checkpointTokenCurrent;

        private CancellationTokenSource _stopRequested = new CancellationTokenSource();

        private Int64 _lastActivityTickCount;

        public void StartAutomaticPolling(
            String checkpointTokenFrom,
            Int32 intervalInMilliseconds,
            Int32 bufferSize = 4000,
            String pollerName = "CommitPollingClient")
        {
            StartManualPolling(checkpointTokenFrom, bufferSize, pollerName);
            _pollerTimer = new System.Timers.Timer(intervalInMilliseconds);
            _pollerTimer.Elapsed += TimerCallback;
            _pollerTimer.Start();

            //Add an immediate poll
            _commandList.Add(command_poll);

            //Set health check for polling
            HealthChecks.RegisterHealthCheck("Polling-" + pollerName, () =>
            {
                if (Status == CommitPollingClientStatus.Stopped)
                {
                    //poller is stopped, system healty
                    return HealthCheckResult.Healthy("Automatic polling stopped");
                }
                else if (Status == CommitPollingClientStatus.Faulted)
                {
                    //poller is stopped, system healty
                    return HealthCheckResult.Unhealthy("Faulted (exception in consumer): " + 
                        LastException != null ? 
                        LastException.ToString() : 
                        "");
                }
                var elapsed = Math.Abs(Environment.TickCount - _lastActivityTickCount);
                if (elapsed > 5000)
                {
                    //more than 5 seconds without a poll, polling probably is stopped
                    return HealthCheckResult.Unhealthy(String.Format("poller stuck, last polling {0} ms ago", elapsed));
                }

                return HealthCheckResult.Healthy("Poller alive");
            });

            Status = CommitPollingClientStatus.Polling;
        }

        public void StartManualPolling(
            String checkpointTokenFrom,
            Int32 bufferSize = 4000,
            String pollerName = "CommitPollingClient")
        {
            _bufferSize = bufferSize;
            _checkpointTokenCurrent = checkpointTokenFrom;
            LastException = null;
            //prepare single poller thread.
            CreateTplChain();
            _pollerThread = new Thread(PollerFunc);
            _pollerThread.IsBackground = true;
            _pollerThread.Name = pollerName;
            _pollerThread.Start();
            MetricsHelper.SetCommitPollingClientBufferSize(pollerName, () => GetClientBufferSize());
        }

        public void StopPolling(Boolean setToFaulted = false)
        {
            if (_pollerTimer != null)
            {
                _pollerTimer.Stop();
                _pollerTimer.Dispose();
                _pollerTimer = null;
            }
            CloseEverything();
            Status = setToFaulted ? 
                CommitPollingClientStatus.Faulted : 
                CommitPollingClientStatus.Stopped;
        }

        private void CloseEverything()
        {
            _stopRequested.Cancel();
            CloseTplChain();
        }

        private void TimerCallback(object sender, System.Timers.ElapsedEventArgs e)
        {
            Poll();
        }

        /// <summary>
        /// Add a manual poll command, it will be executed from the single thread that
        /// read commits from stream.
        /// </summary>
        public void Poll()
        {
            //Add poll command if there are no other poll command in queuel
            if (!_commandList.Any(c => c.Equals(command_poll, StringComparison.OrdinalIgnoreCase)))
                _commandList.Add(command_poll);
        }

        private double GetClientBufferSize()
        {
            if (_buffer != null)
                return _buffer.Count;
            return 0;
        }

{'fraction_non_alphanumeric': 0.0712190719711213, 'fraction_numerical': 0.004361886139730767, 'mean_word_length': 1.7767801211108791, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 37, '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': '5326586', 'n_tokens_mistral': 3358, 'n_tokens_neox': 3157, 'n_words': 765}