<commit_before>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.Practices.TransientFaultHandling;
using NLog;
using ProjectExtensions.Azure.ServiceBus.Serialization;
using System.Threading;
using Microsoft.ServiceBus.Messaging;
using System.IO;
using System.Diagnostics;
using ProjectExtensions.Azure.ServiceBus.TransientFaultHandling.ServiceBus;
using ProjectExtensions.Azure.ServiceBus.Interfaces;

namespace ProjectExtensions.Azure.ServiceBus.Receiver {

    class AzureReceiverHelper {

        static Logger logger = LogManager.GetCurrentClassLogger();

        readonly TopicDescription topic;
        readonly IServiceBusConfigurationFactory configurationFactory;
        readonly IBusConfiguration config;
        readonly RetryPolicy retryPolicy;
        readonly RetryPolicy verifyRetryPolicy;
        AzureBusReceiverState data;
        ServiceBusEnpointData endpoint;
        readonly IServiceBusSerializer serializer;

        int failCounter = 0;

        public AzureBusReceiverState Data {
            get {
                return data;
            }
        }

        public AzureReceiverHelper(TopicDescription topic, IServiceBusConfigurationFactory configurationFactory, IBusConfiguration config,
            IServiceBusSerializer serializer, RetryPolicy verifyRetryPolicy, RetryPolicy retryPolicy, ServiceBusEnpointData endpoint) {
            Guard.ArgumentNotNull(topic, "topic");
            Guard.ArgumentNotNull(configurationFactory, "configurationFactory");
            Guard.ArgumentNotNull(config, "config");
            Guard.ArgumentNotNull(serializer, "serializer");
            Guard.ArgumentNotNull(retryPolicy, "retryPolicy");
            Guard.ArgumentNotNull(endpoint, "endpoint");
            this.topic = topic;
            this.configurationFactory = configurationFactory;
            this.config = config;
            this.serializer = serializer;
            this.verifyRetryPolicy = verifyRetryPolicy;
            this.retryPolicy = retryPolicy;
            this.endpoint = endpoint;

            Configure(endpoint);
        }

        public void ProcessMessagesForSubscription() {
            //ProcessMessagesForSubscriptionLegacy();
            ProcessMessagesForSubscriptionAzure();
        }

        private void ProcessMessagesForSubscriptionAzure() {
            logger.Info("ProcessMessagesForSubscription Message Start {0} Declared {1} MessageTytpe {2}, IsReusable {3}", data.EndPointData.SubscriptionName,
                         data.EndPointData.DeclaredType.ToString(), data.EndPointData.MessageType.ToString(), data.EndPointData.IsReusable);

            var gt = typeof(IReceivedMessage<>).MakeGenericType(data.EndPointData.MessageType);
            var methodInfo = data.EndPointData.DeclaredType.GetMethod("Handle", new Type[] { gt });

            var waitTimeout = TimeSpan.FromSeconds(30);

            // Declare an action implementing the main processing logic for received messages.
            Action<AzureReceiveState> processMessage = ((receiveState) => {

                //For this we must catch any exception
                bool isError = false;
                try {
                    ProcessMessageCallBack(receiveState);
                }
                catch (Exception ex) {
                    //TODO log and whatnot.
                    isError = true;
                }
                finally {
                    try {
                        // With PeekLock mode, we should mark the failed message as abandoned.
                        if (receiveState.Message != null) {
                            if (data.Client.Mode == ReceiveMode.PeekLock) {
                                if (isError) {
                                    // Abandons a brokered message. This will cause Service Bus to unlock the message and make it available 
                                    // to be received again, either by the same consumer or by another completing consumer.
                                    SafeAbandon(receiveState.Message);
                                }
                                else {
                                    // Mark brokered message as completed at which point it's removed from the queue.
                                    SafeComplete(receiveState.Message);
                                }
                            }
                            receiveState.Message.Dispose();
                        }
                    }
                    catch {
                        //Do not fail :)
                    }
                }
            });

            //This should just work

            var options = new OnMessageOptions() {
                AutoComplete = false,
                MaxConcurrentCalls = 20 //Make the app webscale
            };
            options.ExceptionReceived += options_ExceptionReceived;

            //TODO deal with pause when there is an error!!!

            data.Client.OnMessage((msg) => {
                //receive loop begin
                if (msg != null) {
                    // Make sure we are not told to stop receiving while we were waiting for a new message.
                    if (!data.CancelToken.IsCancellationRequested) {
                        // Process the received message.
                        logger.Debug("ProcessMessagesForSubscription Start received new message: {0}", data.EndPointData.SubscriptionNameDebug);
                        var receiveState = new AzureReceiveState(data, methodInfo, serializer, msg);
                        processMessage(receiveState);
                        logger.Debug("ProcessMessagesForSubscription End received new message: {0}", data.EndPointData.SubscriptionNameDebug);
                    }
                }

                if (data.CancelToken.IsCancellationRequested) {
                    //Cancel the message pump.
                    data.SetMessageLoopCompleted();
                    try {
                        data.Client.Close();
                    }
                    catch {
                    }
                }
                //receive loop end
            }, options);
        }

        void options_ExceptionReceived(object sender, ExceptionReceivedEventArgs ex) {

            if (!data.CancelToken.IsCancellationRequested && typeof(ThreadAbortException) != ex.GetType()) {

                //The subscription may have been deleted. If it was, then we want to recreate it.
                var subException = ex.Exception as MessagingEntityNotFoundException;

                if (subException != null && subException.Detail != null && subException.Detail.Message.IndexOf("40400") > -1) {
                    logger.Info("Subscription was deleted. Attempting to Recreate.");
                    Configure(endpoint);
                    logger.Info("Subscription was deleted. Recreated.");
                }
                else {
                    logger.Error(string.Format("ProcessMessagesForSubscription Message Error={0} Declared={1} MessageTytpe={2} IsReusable={3} Error={4}",
                        data.EndPointData.SubscriptionName,<commit_msg>Set up the receiver to recover from a deleted topic. Verified the max calls works. 4 is 4x faster and 5-6x. CPU appears to be the biggest issue.<commit_after>﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.Practices.TransientFaultHandling;
using NLog;
using ProjectExtensions.Azure.ServiceBus.Serialization;
using System.Threading;
using Microsoft.ServiceBus.Messaging;
using System.IO;
using System.Diagnostics;
using ProjectExtensions.Azure.ServiceBus.TransientFaultHandling.ServiceBus;
using ProjectExtensions.Azure.ServiceBus.Interfaces;
using System.Reflection;

namespace ProjectExtensions.Azure.ServiceBus.Receiver {

    class AzureReceiverHelper {

        static Logger logger = LogManager.GetCurrentClassLogger();

        readonly TopicDescription topic;
        readonly IServiceBusConfigurationFactory configurationFactory;
        readonly IBusConfiguration config;
        readonly RetryPolicy retryPolicy;
        readonly RetryPolicy verifyRetryPolicy;
        AzureBusReceiverState data;
        ServiceBusEnpointData endpoint;
        readonly IServiceBusSerializer serializer;
        MethodInfo methodInfo;
        OnMessageOptions options;

        int failCounter = 0;

        public AzureBusReceiverState Data {
            get {
                return data;
            }
        }

        public AzureReceiverHelper(TopicDescription topic, IServiceBusConfigurationFactory configurationFactory, IBusConfiguration config,
            IServiceBusSerializer serializer, RetryPolicy verifyRetryPolicy, RetryPolicy retryPolicy, ServiceBusEnpointData endpoint) {
            Guard.ArgumentNotNull(topic, "topic");
            Guard.ArgumentNotNull(configurationFactory, "configurationFactory");
            Guard.ArgumentNotNull(config, "config");
            Guard.ArgumentNotNull(serializer, "serializer");
            Guard.ArgumentNotNull(retryPolicy, "retryPolicy");
            Guard.ArgumentNotNull(endpoint, "endpoint");
            this.topic = topic;
            this.configurationFactory = configurationFactory;
            this.config = config;
            this.serializer = serializer;
            this.verifyRetryPolicy = verifyRetryPolicy;
            this.retryPolicy = retryPolicy;
            this.endpoint = endpoint;

            Configure(endpoint);
        }

        public void ProcessMessagesForSubscription() {
            var gt = typeof(IReceivedMessage<>).MakeGenericType(data.EndPointData.MessageType);
            methodInfo = data.EndPointData.DeclaredType.GetMethod("Handle", new Type[] { gt });

            options = new OnMessageOptions() {
                AutoComplete = false,
                MaxConcurrentCalls = 8 //Make the app webscale
            };
            options.ExceptionReceived += options_ExceptionReceived;

            //ProcessMessagesForSubscriptionLegacy();
            ProcessMessagesForSubscriptionAzure();
        }

        private bool OnReceiveProcess(AzureReceiveState receiveState) {
            var retVal = false;
            try {
                ProcessMessageCallBack(receiveState);
            }
            catch (Exception ex) {
                //TODO log and whatnot.
                retVal = true;
            }
            finally {
                try {
                    // With PeekLock mode, we should mark the failed message as abandoned.
                    if (receiveState.Message != null) {
                        if (data.Client.Mode == ReceiveMode.PeekLock) {
                            if (retVal) {
                                // Abandons a brokered message. This will cause Service Bus to unlock the message and make it available 
                                // to be received again, either by the same consumer or by another completing consumer.
                                SafeAbandon(receiveState.Message);
                            }
                            else {
                                // Mark brokered message as completed at which point it's removed from the queue.
                                SafeComplete(receiveState.Message);
                            }
                        }
                        receiveState.Message.Dispose();
                    }
                }
                catch {
                    //Do not fail :)
                }
            }
            return retVal;
        }

        public void OnMessageHandler(IBrokeredMessage msg) {
            //receive loop begin
            bool failed = false;
            if (msg != null) {
                // Make sure we are not told to stop receiving while we were waiting for a new message.
                if (!data.CancelToken.IsCancellationRequested) {
                    // Process the received message.
                    logger.Debug("ProcessMessagesForSubscription Start received new message: {0}", data.EndPointData.SubscriptionNameDebug);
                    var receiveState = new AzureReceiveState(data, methodInfo, serializer, msg);
                    failed = OnReceiveProcess(receiveState);
                    logger.Debug("ProcessMessagesForSubscription End received new message: {0}", data.EndPointData.SubscriptionNameDebug);
                }
            }

            if (data.CancelToken.IsCancellationRequested) {
                //Cancel the message pump.
                data.SetMessageLoopCompleted();
                try {
                    data.Client.Close();
                }
                catch {
                }
            }
            else if (failed) {
                //close the pump.
                data.Client.Close();
                if (data.EndPointData.AttributeData.PauseTimeIfErrorWasThrown > 0) {
                    Thread.Sleep(data.EndPointData.AttributeData.PauseTimeIfErrorWasThrown);
                }
                else {
                    Thread.Sleep(1000);
                }
                data.Client.OnMessage(OnMessageHandler, options);
            }
        }

        private void ProcessMessagesForSubscriptionAzure() {
            logger.Info("ProcessMessagesForSubscription Message Start {0} Declared {1} MessageTytpe {2}, IsReusable {3}", data.EndPointData.SubscriptionName,
                         data.EndPointData.DeclaredType.ToString(), data.EndPointData.MessageType.ToString(), data.EndPointData.IsReusable);

            data.Client.OnMessage(OnMessageHandler, options);
        }

        void options_ExceptionReceived(object sender, ExceptionReceivedEventArgs ex) {

            if (!data.CancelToken.IsCancellationRequested && typeof(ThreadAbortException) != ex.GetType()) {

                //The subscription may have been deleted. If it was, then we want to recreate it.
                var subException = ex.Exception as MessagingEntityNotFoundException;

                if (subException != null && subException.Detail != null && subException.Detail.Message.IndexOf("40400") > -1) {
                    logger.Info("Subscription was deleted. Attempting to Recreate.");
                    Configure(endpoint);
                    data.Client.OnMessage(OnMessageHandler, options);
                    logger.Info("Subscription was deleted. Recreated.");
                }
                else {
                    logger.Error(string.Format("ProcessMessagesForSubscription Message Error={0} Declared={1} MessageTytpe={2} IsReusable={3} Error={4}",
                        data.EndPointData.SubscriptionName,
{'fraction_non_alphanumeric': 0.07253780138945648, 'fraction_numerical': 0.0034055305816646235, 'mean_word_length': 1.7688101074863285, 'pattern_counts': {'":': 0, '<': 6, '<?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': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '534400', 'n_tokens_mistral': 3329, 'n_tokens_neox': 3157, 'n_words': 817}