<commit_before>﻿using System;
using ENode.Commanding;
using ENode.Domain;
using ENode.Infrastructure;
using ENode.Infrastructure.Concurrent;
using ENode.Infrastructure.Logging;
using ENode.Infrastructure.Retring;
using ENode.Messaging;
using ENode.Messaging.Impl;

namespace ENode.Eventing.Impl
{
    /// <summary>The default implementation of IUncommittedEventExecutor.
    /// </summary>
    public class DefaultUncommittedEventExecutor : MessageExecutor<EventStream>, IUncommittedEventExecutor
    {
        #region Private Variables

        private readonly IProcessingCommandCache _processingCommandCache;
        private readonly ICommandAsyncResultManager _commandAsyncResultManager;
        private readonly IAggregateRootTypeProvider _aggregateRootTypeProvider;
        private readonly IAggregateRootFactory _aggregateRootFactory;
        private readonly IMemoryCache _memoryCache;
        private readonly IRepository _repository;
        private readonly IRetryCommandService _retryCommandService;
        private readonly IEventStore _eventStore;
        private readonly IEventPublisher _eventPublisher;
        private readonly IRetryService _retryService;
        private readonly IEventPersistenceSynchronizerProvider _eventPersistenceSynchronizerProvider;
        private readonly ILogger _logger;

        #endregion

        #region Constructors

        /// <summary>Parameterized constructor.
        /// </summary>
        /// <param name="processingCommandCache"></param>
        /// <param name="commandAsyncResultManager"></param>
        /// <param name="aggregateRootTypeProvider"></param>
        /// <param name="aggregateRootFactory"></param>
        /// <param name="memoryCache"></param>
        /// <param name="repository"></param>
        /// <param name="retryCommandService"></param>
        /// <param name="eventStore"></param>
        /// <param name="eventPublisher"></param>
        /// <param name="retryService"></param>
        /// <param name="eventPersistenceSynchronizerProvider"></param>
        /// <param name="loggerFactory"></param>
        public DefaultUncommittedEventExecutor(
            IProcessingCommandCache processingCommandCache,
            ICommandAsyncResultManager commandAsyncResultManager,
            IAggregateRootTypeProvider aggregateRootTypeProvider,
            IAggregateRootFactory aggregateRootFactory,
            IMemoryCache memoryCache,
            IRepository repository,
            IRetryCommandService retryCommandService,
            IEventStore eventStore,
            IEventPublisher eventPublisher,
            IRetryService retryService,
            IEventPersistenceSynchronizerProvider eventPersistenceSynchronizerProvider,
            ILoggerFactory loggerFactory)
        {
            _processingCommandCache = processingCommandCache;
            _commandAsyncResultManager = commandAsyncResultManager;
            _aggregateRootTypeProvider = aggregateRootTypeProvider;
            _aggregateRootFactory = aggregateRootFactory;
            _memoryCache = memoryCache;
            _repository = repository;
            _retryCommandService = retryCommandService;
            _eventStore = eventStore;
            _eventPublisher = eventPublisher;
            _retryService = retryService;
            _eventPersistenceSynchronizerProvider = eventPersistenceSynchronizerProvider;
            _logger = loggerFactory.Create(GetType().Name);
        }

        #endregion

        /// <summary>Execute the given event stream.
        /// </summary>
        /// <param name="message"></param>
        /// <param name="queue"></param>
        public override void Execute(EventStream message, IMessageQueue<EventStream> queue)
        {
            var context = new EventStreamContext { EventStream = message, Queue = queue };

            Func<bool> tryCommitEvents = () =>
            {
                try
                {
                    return CommitEvents(context);
                }
                catch (Exception ex)
                {
                    _logger.Error(string.Format("Exception raised when committing events:{0}.", context.EventStream.GetStreamInformation()), ex);
                    return false;
                }
            };

            _retryService.TryAction("TryCommitEvents", tryCommitEvents, 3, () => { });
        }

        #region Private Methods

        private bool CommitEvents(EventStreamContext context)
        {
            var synchronizeResult = TryCallSynchronizersBeforeEventPersisting(context.EventStream);

            switch (synchronizeResult.Status)
            {
                case SynchronizeStatus.SynchronizerConcurrentException:
                    return false;
                case SynchronizeStatus.Failed:
                    Clear(context, synchronizeResult.ErrorInfo);
                    return true;
                default:
                {
                    Action persistSuccessAction = () =>
                    {
                        if (context.HasConcurrentException)
                        {<commit_msg>fix #14, add code to refresh memory before retring command.<commit_after>﻿using System;
using ENode.Commanding;
using ENode.Domain;
using ENode.Infrastructure;
using ENode.Infrastructure.Concurrent;
using ENode.Infrastructure.Logging;
using ENode.Infrastructure.Retring;
using ENode.Messaging;
using ENode.Messaging.Impl;

namespace ENode.Eventing.Impl
{
    /// <summary>The default implementation of IUncommittedEventExecutor.
    /// </summary>
    public class DefaultUncommittedEventExecutor : MessageExecutor<EventStream>, IUncommittedEventExecutor
    {
        #region Private Variables

        private readonly IProcessingCommandCache _processingCommandCache;
        private readonly ICommandAsyncResultManager _commandAsyncResultManager;
        private readonly IAggregateRootTypeProvider _aggregateRootTypeProvider;
        private readonly IAggregateRootFactory _aggregateRootFactory;
        private readonly IMemoryCache _memoryCache;
        private readonly IRepository _repository;
        private readonly IRetryCommandService _retryCommandService;
        private readonly IEventStore _eventStore;
        private readonly IEventPublisher _eventPublisher;
        private readonly IRetryService _retryService;
        private readonly IEventPersistenceSynchronizerProvider _eventPersistenceSynchronizerProvider;
        private readonly ILogger _logger;

        #endregion

        #region Constructors

        /// <summary>Parameterized constructor.
        /// </summary>
        /// <param name="processingCommandCache"></param>
        /// <param name="commandAsyncResultManager"></param>
        /// <param name="aggregateRootTypeProvider"></param>
        /// <param name="aggregateRootFactory"></param>
        /// <param name="memoryCache"></param>
        /// <param name="repository"></param>
        /// <param name="retryCommandService"></param>
        /// <param name="eventStore"></param>
        /// <param name="eventPublisher"></param>
        /// <param name="retryService"></param>
        /// <param name="eventPersistenceSynchronizerProvider"></param>
        /// <param name="loggerFactory"></param>
        public DefaultUncommittedEventExecutor(
            IProcessingCommandCache processingCommandCache,
            ICommandAsyncResultManager commandAsyncResultManager,
            IAggregateRootTypeProvider aggregateRootTypeProvider,
            IAggregateRootFactory aggregateRootFactory,
            IMemoryCache memoryCache,
            IRepository repository,
            IRetryCommandService retryCommandService,
            IEventStore eventStore,
            IEventPublisher eventPublisher,
            IRetryService retryService,
            IEventPersistenceSynchronizerProvider eventPersistenceSynchronizerProvider,
            ILoggerFactory loggerFactory)
        {
            _processingCommandCache = processingCommandCache;
            _commandAsyncResultManager = commandAsyncResultManager;
            _aggregateRootTypeProvider = aggregateRootTypeProvider;
            _aggregateRootFactory = aggregateRootFactory;
            _memoryCache = memoryCache;
            _repository = repository;
            _retryCommandService = retryCommandService;
            _eventStore = eventStore;
            _eventPublisher = eventPublisher;
            _retryService = retryService;
            _eventPersistenceSynchronizerProvider = eventPersistenceSynchronizerProvider;
            _logger = loggerFactory.Create(GetType().Name);
        }

        #endregion

        /// <summary>Execute the given event stream.
        /// </summary>
        /// <param name="message"></param>
        /// <param name="queue"></param>
        public override void Execute(EventStream message, IMessageQueue<EventStream> queue)
        {
            var context = new EventStreamContext { EventStream = message, Queue = queue };

            Func<bool> tryCommitEvents = () =>
            {
                try
                {
                    return CommitEvents(context);
                }
                catch (Exception ex)
                {
                    _logger.Error(string.Format("Exception raised when committing events:{0}.", context.EventStream.GetStreamInformation()), ex);
                    return false;
                }
            };

            _retryService.TryAction("TryCommitEvents", tryCommitEvents, 3, () => { });
        }

        #region Private Methods

        private bool CommitEvents(EventStreamContext context)
        {
            var synchronizeResult = TryCallSynchronizersBeforeEventPersisting(context.EventStream);

            switch (synchronizeResult.Status)
            {
                case SynchronizeStatus.SynchronizerConcurrentException:
                    return false;
                case SynchronizeStatus.Failed:
                    Clear(context, synchronizeResult.ErrorInfo);
                    return true;
                default:
                {
                    Action persistSuccessAction = () =>
                    {
                        if (context.HasConcurrentException)
                        {
                            TryRefreshMemoryCache(context.EventStream);
{'fraction_non_alphanumeric': 0.07598990683229814, 'fraction_numerical': 0.0005822981366459627, 'mean_word_length': 2.318840579710145, 'pattern_counts': {'":': 0, '<': 77, '<?xml version=': 0, '>': 83, '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': '6437498', 'n_tokens_mistral': 2349, 'n_tokens_neox': 2271, 'n_words': 480}