<commit_before>﻿using System.Threading.Tasks;
using ECommon.Logging;
using ECommon.IO;

namespace ENode.Infrastructure.Impl
{
    public abstract class AbstractSequenceProcessingMessageHandler<X, Y, Z> : IProcessingMessageHandler<X, Y, Z>
        where X : class, IProcessingMessage<X, Y, Z>, ISequenceProcessingMessage
        where Y : ISequenceMessage
    {
        #region Private Variables

        private readonly ISequenceMessagePublishedVersionStore _publishedVersionStore;
        private readonly IOHelper _ioHelper;
        private readonly ILogger _logger;

        #endregion

        public abstract string Name { get; }

        #region Constructors

        public AbstractSequenceProcessingMessageHandler(ISequenceMessagePublishedVersionStore publishedVersionStore, IOHelper ioHelper, ILoggerFactory loggerFactory)
        {
            _publishedVersionStore = publishedVersionStore;
            _ioHelper = ioHelper;
            _logger = loggerFactory.Create(GetType().FullName);
        }

        #endregion

        protected abstract Task<AsyncTaskResult> DispatchProcessingMessageAsync(X processingMessage);

        public void HandleAsync(X processingMessage)
        {
            HandleMessageAsync(processingMessage, 0);
        }

        private void HandleMessageAsync(X processingMessage, int retryTimes)
        {
            var message = processingMessage.Message;

            _ioHelper.TryAsyncActionRecursively<AsyncTaskResult<int>>("GetPublishedVersionAsync",
            () => _publishedVersionStore.GetPublishedVersionAsync(Name, message.AggregateRootTypeName, message.AggregateRootStringId),
            currentRetryTimes => HandleMessageAsync(processingMessage, currentRetryTimes),
            result =>
            {
                var publishedVersion = result.Data;
                if (publishedVersion + 1 == message.Version)
                {
                    DispatchProcessingMessageAsync(processingMessage, 0);
                }
                else if (publishedVersion + 1 < message.Version)
                {
                    if (_logger.IsDebugEnabled)
                    {
                        _logger.DebugFormat("The sequence message cannot be process now as the version is not the next version, it will be handle later. contextInfo [aggregateRootId={0},lastPublishedVersion={1},messageVersion={2}]", message.AggregateRootStringId, publishedVersion, message.Version);
                    }
                    processingMessage.AddToWaitingList();
                }
                else
                {
                    processingMessage.Complete(default(Z));
                }
            },
            () => string.Format("sequence message [messageId:{0}, messageType:{1}, aggregateRootId:{2}, aggregateRootVersion:{3}]", message.Id, message.GetType().Name, message.AggregateRootStringId, message.Version),
            null,
            retryTimes,
            true);
        }
        private void DispatchProcessingMessageAsync(X processingMessage, int retryTimes)
        {
            _ioHelper.TryAsyncActionRecursively<AsyncTaskResult>("DispatchProcessingMessageAsync",
            () => DispatchProcessingMessageAsync(processingMessage),
            currentRetryTimes => DispatchProcessingMessageAsync(processingMessage, currentRetryTimes),
            result =>
            {
                UpdatePublishedVersionAsync(processingMessage, 0);
            },
            () => string.Format("sequence message [messageId:{0}, messageType:{1}, aggregateRootId:{2}, aggregateRootVersion:{3}]", processingMessage.Message.Id, processingMessage.Message.GetType().Name, processingMessage.Message.AggregateRootStringId, processingMessage.Message.Version),
            null,
            retryTimes,
            true);<commit_msg>调整一个地方的日志级别<commit_after>﻿using System.Threading.Tasks;
using ECommon.Logging;
using ECommon.IO;

namespace ENode.Infrastructure.Impl
{
    public abstract class AbstractSequenceProcessingMessageHandler<X, Y, Z> : IProcessingMessageHandler<X, Y, Z>
        where X : class, IProcessingMessage<X, Y, Z>, ISequenceProcessingMessage
        where Y : ISequenceMessage
    {
        #region Private Variables

        private readonly ISequenceMessagePublishedVersionStore _publishedVersionStore;
        private readonly IOHelper _ioHelper;
        private readonly ILogger _logger;

        #endregion

        public abstract string Name { get; }

        #region Constructors

        public AbstractSequenceProcessingMessageHandler(ISequenceMessagePublishedVersionStore publishedVersionStore, IOHelper ioHelper, ILoggerFactory loggerFactory)
        {
            _publishedVersionStore = publishedVersionStore;
            _ioHelper = ioHelper;
            _logger = loggerFactory.Create(GetType().FullName);
        }

        #endregion

        protected abstract Task<AsyncTaskResult> DispatchProcessingMessageAsync(X processingMessage);

        public void HandleAsync(X processingMessage)
        {
            HandleMessageAsync(processingMessage, 0);
        }

        private void HandleMessageAsync(X processingMessage, int retryTimes)
        {
            var message = processingMessage.Message;

            _ioHelper.TryAsyncActionRecursively<AsyncTaskResult<int>>("GetPublishedVersionAsync",
            () => _publishedVersionStore.GetPublishedVersionAsync(Name, message.AggregateRootTypeName, message.AggregateRootStringId),
            currentRetryTimes => HandleMessageAsync(processingMessage, currentRetryTimes),
            result =>
            {
                var publishedVersion = result.Data;
                if (publishedVersion + 1 == message.Version)
                {
                    DispatchProcessingMessageAsync(processingMessage, 0);
                }
                else if (publishedVersion + 1 < message.Version)
                {
                    _logger.InfoFormat("The sequence message cannot be process now as the version is not the next version, it will be handle later. contextInfo [aggregateRootId={0},lastPublishedVersion={1},messageVersion={2}]", message.AggregateRootStringId, publishedVersion, message.Version);
                    processingMessage.AddToWaitingList();
                }
                else
                {
                    processingMessage.Complete(default(Z));
                }
            },
            () => string.Format("sequence message [messageId:{0}, messageType:{1}, aggregateRootId:{2}, aggregateRootVersion:{3}]", message.Id, message.GetType().Name, message.AggregateRootStringId, message.Version),
            null,
            retryTimes,
            true);
        }
        private void DispatchProcessingMessageAsync(X processingMessage, int retryTimes)
        {
            _ioHelper.TryAsyncActionRecursively<AsyncTaskResult>("DispatchProcessingMessageAsync",
            () => DispatchProcessingMessageAsync(processingMessage),
            currentRetryTimes => DispatchProcessingMessageAsync(processingMessage, currentRetryTimes),
            result =>
            {
                UpdatePublishedVersionAsync(processingMessage, 0);
            },
            () => string.Format("sequence message [messageId:{0}, messageType:{1}, aggregateRootId:{2}, aggregateRootVersion:{3}]", processingMessage.Message.Id, processingMessage.Message.GetType().Name, processingMessage.Message.AggregateRootStringId, processingMessage.Message.Version),
            null,
            retryTimes,
            true);
{'fraction_non_alphanumeric': 0.07890884896872921, 'fraction_numerical': 0.004258150365934797, 'mean_word_length': 2.6030680728667304, 'pattern_counts': {'":': 0, '<': 19, '<?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': '6126206', 'n_tokens_mistral': 1739, 'n_tokens_neox': 1704, 'n_words': 351}