<commit_before>﻿using IFramework.Command;
using IFramework.Infrastructure;
using IFramework.Infrastructure.Unity.LifetimeManagers;
using IFramework.Message;
using IFramework.MessageQueue.MessageFormat;
using IFramework.SysExceptions;
using IFramework.UnitOfWork;
using Microsoft.ServiceBus.Messaging;
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace IFramework.MessageQueue.ServiceBus
{
    public class CommandConsumer : MessageProcessor, IMessageConsumer
    {
        protected IHandlerProvider _handlerProvider;
        protected Dictionary<string, TopicClient> _replyProducers;
        protected string _commandQueueName;
        protected QueueClient _commandQueueClient;
        protected Task _commandConsumerTask;

        public CommandConsumer(IHandlerProvider handlerProvider,
                               string serviceBusConnectionString,
                               string commandQueueName)
            : base(serviceBusConnectionString)
        {
            _handlerProvider = handlerProvider;
            _commandQueueName = commandQueueName;
            _replyProducers = new Dictionary<string, TopicClient>();
        }


        protected TopicClient GetReplyProducer(string replyTopicName)
        {
            TopicClient replyProducer = null;
            if (!string.IsNullOrWhiteSpace(replyTopicName))
            {
                if (!_replyProducers.TryGetValue(replyTopicName, out replyProducer))
                {
                    try
                    {
                        replyProducer = CreateTopicClient(replyTopicName);
                        _replyProducers[replyTopicName] = replyProducer;
                    }
                    catch (Exception ex)
                    {
                        _logger.Error(ex.GetBaseException().Message, ex);
                    }
                }
            }
            return replyProducer;
        }

        void OnMessageHandled(IMessageContext messageContext, MessageReply reply)
        {
            if (!string.IsNullOrWhiteSpace(messageContext.ReplyToEndPoint) && reply != null)
            {
                var replyProducer = GetReplyProducer(messageContext.ReplyToEndPoint);
                if (replyProducer != null)
                {
                    while (true)
                    {
                        try
                        {
                            replyProducer.Send(reply.BrokeredMessage);
                            break;
                        }
                        catch (Exception)
                        {
                            Thread.Sleep(1000);
                        }
                    }
                    _logger.InfoFormat("send reply, commandID:{0}", reply.MessageID);
                }
            }
        }

        public void Start()
        {
            try
            {
                _commandQueueClient = CreateQueueClient(_commandQueueName);
                _commandConsumerTask = Task.Factory.StartNew(ConsumeMessages, TaskCreationOptions.LongRunning);

            }
            catch (Exception ex)
            {
                _logger.Error(ex.GetBaseException().Message, ex);
            }
        }

        protected virtual void ConsumeMessages()
        {
            while (!_exit)
            {
                try
                {
                    BrokeredMessage brokeredMessage = null;
                    brokeredMessage = _commandQueueClient.Receive();
                    if (brokeredMessage != null)
                    {
                        ConsumeMessage(brokeredMessage);
                        brokeredMessage.Complete();
                        MessageCount++;
                    }
                }
                catch (Exception ex)
                {
                    Thread.Sleep(1000);
                    _logger.Error(ex.GetBaseException().Message, ex);
                }
            }
        }

        void ConsumeMessage(BrokeredMessage brokeredMessage)
        {
            var commandContext = new MessageContext(brokeredMessage);
            var message = commandContext.Message as ICommand;
            MessageReply messageReply = null;
            if (message == null)
            {
                return;
            }
            var needRetry = message.NeedRetry;
            bool commandHasHandled = false;
            IMessageStore messageStore = null;
            do
            {
                try
                {
                    PerMessageContextLifetimeManager.CurrentMessageContext = commandContext;
                    messageStore = IoCFactory.Resolve<IMessageStore>();
                    commandHasHandled = messageStore.HasCommandHandled(commandContext.MessageID);
                    if (!commandHasHandled)
                    {
                        var messageHandler = _handlerProvider.GetHandler(message.GetType());
                        _logger.InfoFormat("Handle command, commandID:{0}", commandContext.MessageID);

                        if (messageHandler == null)
                        {
                            messageReply = new MessageReply(commandContext.MessageID, new NoHandlerExists());
                        }
                        else
                        {
                            var unitOfWork = IoCFactory.Resolve<IUnitOfWork>();
                            ((dynamic)messageHandler).Handle((dynamic)message);
                            unitOfWork.Commit();
                            messageReply = new MessageReply(commandContext.MessageID, commandContext.Reply);
                        }
                    }
                    needRetry = false;
                }
                catch (Exception e)
                {
                    if (!(e is OptimisticConcurrencyException) || !needRetry)
                    {
                        messageReply = new MessageReply(commandContext.MessageID, e.GetBaseException());
                        if (e is DomainException)
                        {
                            _logger.Warn(message.ToJson(), e);
                        }
                        else
                        {
                            _logger.Error(message.ToJson(), e);
                        }
                        if (messageStore != null)
                        {
                            messageStore.SaveFailedCommand(commandContext);
                        }
                        needRetry = false;
                    }
                }
                finally
                {
                    PerMessageContextLifetimeManager.CurrentMessageContext = null;
                }
            } while (needRetry);
            if (!commandHasHandled)
            {
                OnMessageHandled(commandContext, messageReply);
            }
        }


        public void Stop()
        {
            _exit = true;
            if (_commandConsumerTask != null)
            {
                _commandQueueClient.Close();
                if (!_commandConsumerTask.Wait(1000))
                {
                    _logger.ErrorFormat("receiver can't be stopped!");
                }
            }
        }

        public string GetStatus()
        {
            return this.ToString();
        }

        public decimal MessageCount { get; set; }
    }
}
<commit_msg>add log for send message error<commit_after>﻿using IFramework.Command;
using IFramework.Infrastructure;
using IFramework.Infrastructure.Unity.LifetimeManagers;
using IFramework.Message;
using IFramework.MessageQueue.MessageFormat;
using IFramework.SysExceptions;
using IFramework.UnitOfWork;
using Microsoft.ServiceBus.Messaging;
using System;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace IFramework.MessageQueue.ServiceBus
{
    public class CommandConsumer : MessageProcessor, IMessageConsumer
    {
        protected IHandlerProvider _handlerProvider;
        protected Dictionary<string, TopicClient> _replyProducers;
        protected string _commandQueueName;
        protected QueueClient _commandQueueClient;
        protected Task _commandConsumerTask;

        public CommandConsumer(IHandlerProvider handlerProvider,
                               string serviceBusConnectionString,
                               string commandQueueName)
            : base(serviceBusConnectionString)
        {
            _handlerProvider = handlerProvider;
            _commandQueueName = commandQueueName;
            _replyProducers = new Dictionary<string, TopicClient>();
        }


        protected TopicClient GetReplyProducer(string replyTopicName)
        {
            TopicClient replyProducer = null;
            if (!string.IsNullOrWhiteSpace(replyTopicName))
            {
                if (!_replyProducers.TryGetValue(replyTopicName, out replyProducer))
                {
                    try
                    {
                        replyProducer = CreateTopicClient(replyTopicName);
                        _replyProducers[replyTopicName] = replyProducer;
                    }
                    catch (Exception ex)
                    {
                        _logger.Error(ex.GetBaseException().Message, ex);
                    }
                }
            }
            return replyProducer;
        }

        void OnMessageHandled(IMessageContext messageContext, MessageReply reply)
        {
            if (!string.IsNullOrWhiteSpace(messageContext.ReplyToEndPoint) && reply != null)
            {
                var replyProducer = GetReplyProducer(messageContext.ReplyToEndPoint);
                if (replyProducer != null)
                {
                    while (true)
                    {
                        try
                        {
                            replyProducer.Send(reply.BrokeredMessage);
                            break;
                        }
                        catch (Exception ex)
                        {
                            Thread.Sleep(1000);
                            _logger.Error(ex.GetBaseException().Message, ex);
                        }
                    }
                    _logger.InfoFormat("send reply, commandID:{0}", reply.MessageID);
                }
            }
        }

        public void Start()
        {
            try
            {
                _commandQueueClient = CreateQueueClient(_commandQueueName);
                _commandConsumerTask = Task.Factory.StartNew(ConsumeMessages, TaskCreationOptions.LongRunning);

            }
            catch (Exception ex)
            {
                _logger.Error(ex.GetBaseException().Message, ex);
            }
        }

        protected virtual void ConsumeMessages()
        {
            while (!_exit)
            {
                try
                {
                    BrokeredMessage brokeredMessage = null;
                    brokeredMessage = _commandQueueClient.Receive();
                    if (brokeredMessage != null)
                    {
                        ConsumeMessage(brokeredMessage);
                        brokeredMessage.Complete();
                        MessageCount++;
                    }
                }
                catch (Exception ex)
                {
                    Thread.Sleep(1000);
                    _logger.Debug(ex.GetBaseException().Message, ex);
                }
            }
        }

        void ConsumeMessage(BrokeredMessage brokeredMessage)
        {
            var commandContext = new MessageContext(brokeredMessage);
            var message = commandContext.Message as ICommand;
            MessageReply messageReply = null;
            if (message == null)
            {
                return;
            }
            var needRetry = message.NeedRetry;
            bool commandHasHandled = false;
            IMessageStore messageStore = null;
            do
            {
                try
                {
                    PerMessageContextLifetimeManager.CurrentMessageContext = commandContext;
                    messageStore = IoCFactory.Resolve<IMessageStore>();
                    commandHasHandled = messageStore.HasCommandHandled(commandContext.MessageID);
                    if (!commandHasHandled)
                    {
                        var messageHandler = _handlerProvider.GetHandler(message.GetType());
                        _logger.InfoFormat("Handle command, commandID:{0}", commandContext.MessageID);

                        if (messageHandler == null)
                        {
                            messageReply = new MessageReply(commandContext.MessageID, new NoHandlerExists());
                        }
                        else
                        {
                            var unitOfWork = IoCFactory.Resolve<IUnitOfWork>();
                            ((dynamic)messageHandler).Handle((dynamic)message);
                            unitOfWork.Commit();
                            messageReply = new MessageReply(commandContext.MessageID, commandContext.Reply);
                        }
                    }
                    needRetry = false;
                }
                catch (Exception e)
                {
                    if (!(e is OptimisticConcurrencyException) || !needRetry)
                    {
                        messageReply = new MessageReply(commandContext.MessageID, e.GetBaseException());
                        if (e is DomainException)
                        {
                            _logger.Warn(message.ToJson(), e);
                        }
                        else
                        {
                            _logger.Error(message.ToJson(), e);
                        }
                        if (messageStore != null)
                        {
                            messageStore.SaveFailedCommand(commandContext);
                        }
                        needRetry = false;
                    }
                }
                finally
                {
                    PerMessageContextLifetimeManager.CurrentMessageContext = null;
                }
            } while (needRetry);
            if (!commandHasHandled)
            {
                OnMessageHandled(commandContext, messageReply);
            }
        }


        public void Stop()
        {
            _exit = true;
            if (_commandConsumerTask != null)
            {
                _commandQueueClient.Close();
                if (!_commandConsumerTask.Wait(1000))
                {
                    _logger.ErrorFormat("receiver can't be stopped!");
                }
            }
        }

        public string GetStatus()
        {
            return this.ToString();
        }

        public decimal MessageCount { get; set; }
    }
}

{'fraction_non_alphanumeric': 0.06544112887565538, 'fraction_numerical': 0.0018124150430448573, 'mean_word_length': 1.1103674361426035, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, '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': '5014625', 'n_tokens_mistral': 3389, 'n_tokens_neox': 3143, 'n_words': 578}