<commit_before>﻿using Castle.Core.Logging;
using Jarvis.Framework.Kernel.Support;
using NEventStore;
using NEventStore.Client;
using NEventStore.Persistence;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
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;

        readonly ILogger _logger;
        private IPersistStreams _persistStreams;

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

        public void AddConsumer(IPollingClientConsumer consumer)
        {
            ExecutionDataflowBlockOptions consumerOptions = new ExecutionDataflowBlockOptions();
            consumerOptions.BoundedCapacity = _bufferSize;
            _consumers.Add(new ActionBlock<ICommit>(
                  commit =>
                  {
                      consumer.Consume(commit);
                  }, consumerOptions));
        }

        public void AddConsumer(Action<ICommit> consumerAction)
        {
            ExecutionDataflowBlockOptions consumerOptions = new ExecutionDataflowBlockOptions();
            consumerOptions.BoundedCapacity = _bufferSize;
            _consumers.Add(new ActionBlock<ICommit>(consumerAction, consumerOptions));
        }

        #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();

        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);
        }

        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()
        {
            //TODO: Decide if we want to wait TPL
            _pollerTimer.Stop();
            _pollerTimer.Dispose();

            _stopRequested.Cancel();
            _commandList.Add(command_stop);
        }

        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;
        }

        private void PollerFunc(object obj)
        {
            foreach (var command in _commandList.GetConsumingEnumerable())
            {
                try
                {
                    switch (command)
                    {
                        case command_poll:
                            InternalPoll();
                            break;

                        case command_stop:
                            CloseTplChain();
                            return; //this will stop poller thread

                        default:
                            _logger.ErrorFormat("Unknown command {0}", command);
                            break;
                    }
                }
                catch (Exception ex)
                {
                    _logger.ErrorFormat(ex, "Error in executing command: {0}", command);
                    throw;
                }
            }
        }

        private void InternalPoll()
        {
            if (_persistStreams.IsDisposed) return; //no need to poll, someone disposed NEventStore
            try
            {
                IEnumerable<ICommit> commits = _persistStreams.GetFrom(_checkpointTokenCurrent);
                foreach (var commit in commits)
                {
                    if (_stopRequested.IsCancellationRequested)
                    {
                        CloseTplChain();
                        return;
                    }

                    _enhancer.Enhance(commit);
                    while (!_buffer.SendAsync(commit).Wait(2000))
                    {
                        //maybe the mesh is full, but check for completion
                        if (_stopRequested.IsCancellationRequested)
                        {
                            _buffer.Complete();
                            return;
                        }
                    };
                    _checkpointTokenCurrent = commit.CheckpointToken;
                }
            }
            catch (Exception ex)
            {
                if (_stopRequested.IsCancellationRequested)
                    return;

                // These exceptions are expected to be transient, we can simply start a new poll immediately
                _logger.ErrorFormat(ex, "Error in polling client {0}", ex.Message);

                Poll();
            }
        }

        private void CloseTplChain()
        {
            //signal Tpl to complete buffer.<commit_msg>Fixed new commit poller with TPL<commit_after>﻿using Castle.Core.Logging;
using Jarvis.Framework.Kernel.Support;
using NEventStore;
using NEventStore.Client;
using NEventStore.Persistence;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
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;

        readonly ILogger _logger;
        private IPersistStreams _persistStreams;

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

        public void AddConsumer(IPollingClientConsumer consumer)
        {
            ExecutionDataflowBlockOptions consumerOptions = new ExecutionDataflowBlockOptions();
            consumerOptions.BoundedCapacity = _bufferSize;
            _consumers.Add(new ActionBlock<ICommit>(
                  commit =>
                  {
                      consumer.Consume(commit);
                  }, consumerOptions));
        }

        public void AddConsumer(Action<ICommit> consumerAction)
        {
            ExecutionDataflowBlockOptions consumerOptions = new ExecutionDataflowBlockOptions();
            consumerOptions.BoundedCapacity = _bufferSize;
            _consumers.Add(new ActionBlock<ICommit>(consumerAction, consumerOptions));
        }

        #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();

        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);
        }

        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()
        {
            //TODO: Decide if we want to wait TPL
            _pollerTimer.Stop();
            _pollerTimer.Dispose();

            _stopRequested.Cancel();
            _commandList.Add(command_stop);
        }

        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;
        }

        private void PollerFunc(object obj)
        {
            foreach (var command in _commandList.GetConsumingEnumerable())
            {
                try
                {
                    switch (command)
                    {
                        case command_poll:
                            InternalPoll();
                            break;

                        case command_stop:
                            CloseTplChain();
                            return; //this will stop poller thread

                        default:
                            _logger.ErrorFormat("Unknown command {0}", command);
                            break;
                    }
                }
                catch (Exception ex)
                {
                    _logger.ErrorFormat(ex, "Error in executing command: {0}", command);
                    throw;
                }
            }
        }

        private void InternalPoll()
        {
            if (_persistStreams.IsDisposed) return; //no need to poll, someone disposed NEventStore
            try
            {
                IEnumerable<ICommit> commits = _persistStreams.GetFrom(_checkpointTokenCurrent);
                foreach (var commit in commits)
                {
                    if (_stopRequested.IsCancellationRequested)
                    {
                        CloseTplChain();
                        return;
                    }

                    _enhancer.Enhance(commit);
                    var task = _buffer.SendAsync(commit);
                    while (!task.Wait(2000))
                    {
                        //maybe the mesh is full, but check for completion
                        if (_stopRequested.IsCancellationRequested)
                        {
                            _buffer.Complete();
                            return;
                        }
                    };
                    _checkpointTokenCurrent = commit.CheckpointToken;
                }
            }
            catch (Exception ex)
            {
                if (_stopRequested.IsCancellationRequested)
                    return;

                // These exceptions are expected to be transient, we can simply start a new poll immediately
                _logger.ErrorFormat(ex, "Error in polling client {0}", ex.Message);

                Poll();
            }
        }

        private void CloseTplChain()
        {
            //signal Tpl to complete buffer.
{'fraction_non_alphanumeric': 0.0671251719394773, 'fraction_numerical': 0.003576341127922971, 'mean_word_length': 1.5218522372528616, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 33, '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': '7322122', 'n_tokens_mistral': 3529, 'n_tokens_neox': 3335, 'n_words': 797}