<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Aggregates.Exceptions;
using Aggregates.Messages;
using NServiceBus;
using NServiceBus.ObjectBuilder;
using NServiceBus.Pipeline;
using NServiceBus.Pipeline.Contexts;
using NServiceBus.Logging;
using Metrics;
using Aggregates.Extensions;
using Newtonsoft.Json;
using NServiceBus.Settings;

namespace Aggregates.Internal
{
    internal class ExceptionRejector : IBehavior<IncomingContext>
    {
        private static readonly ILog Logger = LogManager.GetLogger(typeof(ExceptionRejector));

        private static Meter _errorsMeter = Metric.Meter("Message Faults", Unit.Errors);
        private readonly IBus _bus;
        private readonly ReadOnlySettings _settings;
        private readonly Int32 _maxRetries;

        public ExceptionRejector(IBus bus, ReadOnlySettings settings)
        {
            _bus = bus;
            _settings = settings;
            _maxRetries = _settings.Get<Int32>("MaxRetries");
        }

        public void Invoke(IncomingContext context, Action next)
        {
            try
            {
                next();
            }
            catch (Exception e)
            {
                var numberOfRetries = GetNumberOfFirstLevelRetries(context.PhysicalMessage) + 1;
                context.PhysicalMessage.Headers[Headers.Retries] = numberOfRetries.ToString();
                if (numberOfRetries < _maxRetries)
                {
                    Logger.WarnFormat("Message {2} type {0} has faulted! {1} times", context.IncomingLogicalMessage.MessageType.FullName, GetNumberOfFirstLevelRetries(context.PhysicalMessage), context.PhysicalMessage.Id);
                    throw;
                }
                

                _errorsMeter.Mark();
                try
                {
                    Logger.WarnFormat("Message {2} type {0} has faulted!\nException: {1}", context.IncomingLogicalMessage.MessageType.FullName, e, context.PhysicalMessage.Id);
                }
                catch (KeyNotFoundException)
                {
                    Logger.WarnFormat("Message {1} [Unknown] has faulted!\nException: {0}", e, context.PhysicalMessage.Id);
                }
                // Tell the sender the command was not handled due to a service exception
                var rejection = context.Builder.Build<Func<Exception, String, Error>>();
                // Wrap exception in our object which is serializable
                _bus.Reply(rejection(e, $"Rejected message {context.IncomingLogicalMessage.MessageType.FullName}\n Payload: {JsonConvert.SerializeObject(context.IncomingLogicalMessage.Instance)}"));
            }

        }
        static int GetNumberOfFirstLevelRetries(TransportMessage message)
        {
            string value;
            if (message.Headers.TryGetValue(Headers.Retries, out value))
            {
                int i;
                if (int.TryParse(value, out i))
                {
                    return i;
                }
            }
            return 0;
        }
    }

    internal class ExceptionRejectorRegistration : RegisterStep
    {
        public ExceptionRejectorRegistration()
            : base("ExceptionRejector", typeof(ExceptionRejector), "Catches exceptions thrown while processing and reports to client via IReject")
        {
            InsertBefore(WellKnownStep.CreateChildContainer);

        }
    }
}
<commit_msg>Need to handle retry count myself - NSB's first level retries just dont work<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Aggregates.Exceptions;
using Aggregates.Messages;
using NServiceBus;
using NServiceBus.ObjectBuilder;
using NServiceBus.Pipeline;
using NServiceBus.Pipeline.Contexts;
using NServiceBus.Logging;
using Metrics;
using Aggregates.Extensions;
using Newtonsoft.Json;
using NServiceBus.Settings;

namespace Aggregates.Internal
{
    internal class ExceptionRejector : IBehavior<IncomingContext>
    {
        private static readonly ILog Logger = LogManager.GetLogger(typeof(ExceptionRejector));

        private static IDictionary<String, Int32> _retryRegistry = new Dictionary<String, Int32>();
        private static Meter _errorsMeter = Metric.Meter("Message Faults", Unit.Errors);
        private readonly IBus _bus;
        private readonly ReadOnlySettings _settings;
        private readonly Int32 _maxRetries;

        public ExceptionRejector(IBus bus, ReadOnlySettings settings)
        {
            _bus = bus;
            _settings = settings;
            _maxRetries = _settings.Get<Int32>("MaxRetries");
        }

        public void Invoke(IncomingContext context, Action next)
        {
            var messageId = context.PhysicalMessage.Id;
            try
            {
                if(_retryRegistry.ContainsKey(messageId))
                    context.PhysicalMessage.Headers[Headers.Retries] = _retryRegistry[messageId].ToString();

                next();
                _retryRegistry.Remove(context.PhysicalMessage.Id);
            }
            catch (Exception e)
            {
                var numberOfRetries = 0;
                _retryRegistry.TryGetValue(messageId, out numberOfRetries);
                    
                if (numberOfRetries < _maxRetries)
                {
                    Logger.WarnFormat("Message {2} type {0} has faulted! {1} times", context.IncomingLogicalMessage.MessageType.FullName, numberOfRetries, context.PhysicalMessage.Id);
                    _retryRegistry[messageId] = numberOfRetries + 1;
                    throw;
                }
                _retryRegistry.Remove(messageId);


                _errorsMeter.Mark();
                try
                {
                    Logger.WarnFormat("Message {2} type {0} has faulted!\nException: {1}", context.IncomingLogicalMessage.MessageType.FullName, e, context.PhysicalMessage.Id);
                }
                catch (KeyNotFoundException)
                {
                    Logger.WarnFormat("Message {1} [Unknown] has faulted!\nException: {0}", e, context.PhysicalMessage.Id);
                }
                // Tell the sender the command was not handled due to a service exception
                var rejection = context.Builder.Build<Func<Exception, String, Error>>();
                // Wrap exception in our object which is serializable
                _bus.Reply(rejection(e, $"Rejected message {context.IncomingLogicalMessage.MessageType.FullName}\n Payload: {JsonConvert.SerializeObject(context.IncomingLogicalMessage.Instance)}"));
            }

        }
        static int GetNumberOfFirstLevelRetries(TransportMessage message)
        {
            Int32 value;
            if (_retryRegistry.TryGetValue(message.Id, out value))
            {
                return value;
            }
            return 0;
        }
    }

    internal class ExceptionRejectorRegistration : RegisterStep
    {
        public ExceptionRejectorRegistration()
            : base("ExceptionRejector", typeof(ExceptionRejector), "Catches exceptions thrown while processing and reports to client via IReject")
        {
            InsertBefore(WellKnownStep.CreateChildContainer);

        }
    }
}

{'fraction_non_alphanumeric': 0.08003266639444671, 'fraction_numerical': 0.004763849190145638, 'mean_word_length': 2.1686071582578696, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 13, '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': '5359', 'n_tokens_mistral': 1896, 'n_tokens_neox': 1818, 'n_words': 425}