<commit_before>﻿namespace King.Service.ServiceFabric
{
    using Azure.Data;
    using Microsoft.ServiceFabric.Data.Collections;
    using Microsoft.ServiceFabric.Services.Runtime;
    using System;
    using System.Diagnostics;
    using System.Threading;
    using System.Threading.Tasks;

    /// <summary>
    /// Dequeue Service
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class DequeueService<T> : StatefulService
    {
        #region Members
        /// <summary>
        /// Processor
        /// </summary>
        protected readonly IProcessor<T> processor;

        /// <summary>
        /// Queue Name
        /// </summary>
        protected readonly string queueName;
        #endregion

        #region Constructors
        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="queueName">Queue Name</param>
        /// <param name="processor">Processor</param>
        public DequeueService(string queueName, IProcessor<T> processor)
        {
            if (string.IsNullOrWhiteSpace(queueName))
            {
                throw new ArgumentException("queueName");
            }
            if (null == processor)
            {
                throw new ArgumentNullException("processor");
            }

            this.processor = processor;
            this.queueName = queueName;
        }
        #endregion

        #region Methods
        /// <summary>
        /// Run Async
        /// </summary>
        /// <param name="cancellationToken">Cancellation Token</param>
        /// <returns>Task</returns>
        protected override async Task RunAsync(CancellationToken cancellationToken)
        {
            if (!cancellationToken.IsCancellationRequested)
            {
                var queue = await this.StateManager.GetOrAddAsync<IReliableQueue<T>>(this.queueName);

                using (var tx = this.StateManager.CreateTransaction())
                {
                    var message = await queue.TryDequeueAsync(tx);

                    if (message.HasValue)
                    {
                        var success = await this.processor.Process(message.Value);

                        if (success)
                        {
                            await tx.CommitAsync();
                        }
                    }
                }
            }
            else
            {
                Trace.TraceInformation("Dequeue not run, cancelation called.");
            }
        }
        #endregion
    }
}<commit_msg>Switch to loop and control speed<commit_after>﻿namespace King.Service.ServiceFabric
{
    using Azure.Data;
    using Microsoft.ServiceFabric.Data.Collections;
    using Microsoft.ServiceFabric.Services.Runtime;
    using System;
    using System.Diagnostics;
    using System.Threading;
    using System.Threading.Tasks;

    /// <summary>
    /// Dequeue Service
    /// </summary>
    /// <typeparam name="T"></typeparam>
    public class DequeueService<T> : StatefulService
    {
        #region Members
        /// <summary>
        /// Processor
        /// </summary>
        protected readonly IProcessor<T> processor;

        /// <summary>
        /// Queue Name
        /// </summary>
        protected readonly string queueName;

        protected readonly int seconds = 15;
        #endregion

        #region Constructors
        /// <summary>
        /// Constructor
        /// </summary>
        /// <param name="queueName">Queue Name</param>
        /// <param name="processor">Processor</param>
        /// <param name="seconds">Check every</param>
        public DequeueService(string queueName, IProcessor<T> processor, int seconds = 15)
        {
            if (string.IsNullOrWhiteSpace(queueName))
            {
                throw new ArgumentException("queueName");
            }
            if (null == processor)
            {
                throw new ArgumentNullException("processor");
            }

            this.processor = processor;
            this.queueName = queueName;
            this.seconds = 0 > seconds ? 15 : seconds;
        }
        #endregion

        #region Methods
        /// <summary>
        /// Run Async
        /// </summary>
        /// <param name="cancellationToken">Cancellation Token</param>
        /// <returns>Task</returns>
        protected override async Task RunAsync(CancellationToken cancellationToken)
        {
            try
            {
                while (!cancellationToken.IsCancellationRequested)
                {
                    var queue = await this.StateManager.GetOrAddAsync<IReliableQueue<T>>(this.queueName);

                    using (var tx = this.StateManager.CreateTransaction())
                    {
                        var message = await queue.TryDequeueAsync(tx);

                        if (message.HasValue)
                        {
                            var success = await this.processor.Process(message.Value);

                            if (success)
                            {
                                await tx.CommitAsync();
                            }
                        }
                    }

                    await Task.Delay(TimeSpan.FromSeconds(this.seconds), cancellationToken);
                }
            }
            catch (TaskCanceledException ex)
            {
                Trace.TraceError("Task Canceled Exception, can be normal: {0}", ex);
            }
        }
        #endregion
    }
}
{'fraction_non_alphanumeric': 0.10261532873229204, 'fraction_numerical': 0.001452960406828914, 'mean_word_length': 1.3788336933045355, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 56, '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': '7649430', 'n_tokens_mistral': 1340, 'n_tokens_neox': 1288, 'n_words': 274}