<commit_before>﻿namespace MassTransit.AzureServiceBusTransport
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    using Azure.Messaging.ServiceBus;
    using Azure.Messaging.ServiceBus.Administration;
    using Internals;
    using MassTransit.Middleware;
    using Transports;


    public class ServiceBusConnectionContext :
        BasePipeContext,
        ConnectionContext,
        IAsyncDisposable
    {
        readonly ServiceBusAdministrationClient _administrationClient;
        readonly ServiceBusClient _client;

        public ServiceBusConnectionContext(ServiceBusClient client, ServiceBusAdministrationClient administrationClient, CancellationToken cancellationToken)
            : base(cancellationToken)
        {
            _client = client;
            _administrationClient = administrationClient;
            Endpoint = new Uri($"sb://{_client.FullyQualifiedNamespace}");
        }

        public Uri Endpoint { get; }

        public ServiceBusProcessor CreateQueueProcessor(ReceiveSettings settings)
        {
            return _client.CreateProcessor(settings.Path, GetProcessorOptions(settings));
        }

        public ServiceBusSessionProcessor CreateQueueSessionProcessor(ReceiveSettings settings)
        {
            return _client.CreateSessionProcessor(settings.Path, GetSessionProcessorOptions(settings));
        }

        public ServiceBusProcessor CreateSubscriptionProcessor(SubscriptionSettings settings)
        {
            return _client.CreateProcessor(settings.CreateTopicOptions.Name, settings.CreateSubscriptionOptions.SubscriptionName,
                GetProcessorOptions(settings));
        }

        public ServiceBusSessionProcessor CreateSubscriptionSessionProcessor(SubscriptionSettings settings)
        {
            return _client.CreateSessionProcessor(settings.CreateTopicOptions.Name, settings.CreateSubscriptionOptions.SubscriptionName,
                GetSessionProcessorOptions(settings));
        }

        public ServiceBusSender CreateMessageSender(string entityPath)
        {
            return _client.CreateSender(entityPath);
        }

        public async Task<QueueProperties> CreateQueue(CreateQueueOptions createQueueOptions)
        {
            QueueProperties queueProperties = null;
            try
            {
                queueProperties = await GetQueueAsync(createQueueOptions.Name).ConfigureAwait(false);
            }
            catch (ServiceBusException exception) when (exception.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
            {
            }

            if (queueProperties == null)
            {
                try
                {
                    TransportLogMessages.CreateQueue(createQueueOptions.Name);

                    queueProperties = await CreateQueueAsync(createQueueOptions).ConfigureAwait(false);
                }
                catch (ServiceBusException sbe) when (sbe.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
                {
                    queueProperties = await GetQueueAsync(createQueueOptions.Name).ConfigureAwait(false);
                }
            }

            LogContext.Debug?.Log("Queue: {Queue} ({Attributes})", createQueueOptions.Name,
                string.Join(", ",
                    new[]
                    {
                        createQueueOptions.RequiresDuplicateDetection ? "dupe detect" : "",
                        createQueueOptions.DeadLetteringOnMessageExpiration ? "dead letter" : "",
                        createQueueOptions.RequiresSession ? "session" : "",
                        createQueueOptions.AutoDeleteOnIdle != Defaults.AutoDeleteOnIdle
                            ? $"auto-delete: {createQueueOptions.AutoDeleteOnIdle.ToFriendlyString()}"
                            : ""
                    }.Where(x => !string.IsNullOrWhiteSpace(x))));

            return queueProperties;
        }

        public async Task<TopicProperties> CreateTopic(CreateTopicOptions createTopicOptions)
        {
            TopicProperties topicProperties = null;
            try
            {
                topicProperties = await GetTopicAsync(createTopicOptions.Name).ConfigureAwait(false);
            }
            catch (ServiceBusException exception) when (exception.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
            {
            }

            if (topicProperties == null)
            {
                try
                {
                    TransportLogMessages.CreateTopic(createTopicOptions.Name);

                    topicProperties = await CreateTopicAsync(createTopicOptions).ConfigureAwait(false);
                }
                catch (ServiceBusException e) when (e.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
                {
                    topicProperties = await GetTopicAsync(createTopicOptions.Name).ConfigureAwait(false);
                }
            }

            LogContext.Debug?.Log("Topic: {Topic} ({Attributes})", createTopicOptions.Name,
                string.Join(", ",
                    new[]
                    {
                        createTopicOptions.RequiresDuplicateDetection ? "dupe detect" : "",
                        createTopicOptions.EnablePartitioning ? "partitioned" : "",
                        createTopicOptions.SupportOrdering ? "ordered" : ""
                    }.Where(x => !string.IsNullOrWhiteSpace(x))));

            return topicProperties;
        }

        public async Task<SubscriptionProperties> CreateTopicSubscription(CreateSubscriptionOptions createSubscriptionOptions, CreateRuleOptions rule,
            RuleFilter filter)
        {
            var create = true;
            SubscriptionProperties subscriptionProperties = null;
            try
            {
                subscriptionProperties = await GetSubscriptionAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName)
                    .ConfigureAwait(false);

                string NormalizeForwardTo(string forwardTo)
                {
                    return string.IsNullOrEmpty(forwardTo)
                        ? string.Empty
                        : forwardTo.Replace(Endpoint.ToString(), string.Empty).Trim('/');
                }

                var targetForwardTo = NormalizeForwardTo(createSubscriptionOptions.ForwardTo);
                var currentForwardTo = NormalizeForwardTo(subscriptionProperties.ForwardTo);

                if (!targetForwardTo.Equals(currentForwardTo))
                {
                    LogContext.Debug?.Log("Updating subscription: {Subscription} ({Topic} -> {ForwardTo})", subscriptionProperties.SubscriptionName,
                        subscriptionProperties.TopicName, subscriptionProperties.ForwardTo);

                    await UpdateSubscriptionAsync(subscriptionProperties).ConfigureAwait(false);
                }

                if (rule != null)
                {
                    var ruleProperties = await GetRuleAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName, rule.Name)
                        .ConfigureAwait(false);
                    if (rule.Name == ruleProperties.Name && (!rule.Filter.Equals(ruleProperties.Filter) || !rule.Action.Equals(ruleProperties.Action)))
                    {
                        LogContext.Debug?.Log("Updating subscription Rule: {Rule} ({DescriptionFilter} -> {Filter})", rule.Name,
                            ruleProperties.Filter.ToString(), rule.Filter.ToString());

                        ruleProperties.Filter = rule.Filter;
                        ruleProperties.Action = rule.Action;

                        await UpdateRuleAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName, ruleProperties)
                            .ConfigureAwait(false);
                    }
                }
                else if (filter != null)
                {
                    IList<RuleProperties> rules = await GetRulesAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName)
                        .ConfigureAwait(false);
                    if (rules.Count == 1)
                    {
                        var existingRule = rules[0];

                        if (Guid.TryParse(existingRule.Name, out _) && !existingRule.Filter.Equals(filter))
                        {
                            LogContext.Debug?.Log("Updating subscription filter: {Rule} ({DescriptionFilter} -> {Filter})", existingRule.Name,
                                existingRule.Filter.ToString(), filter.ToString());

                            existingRule.Filter = filter;

                            await UpdateRuleAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName, existingRule)
                                .ConfigureAwait(false);
                        }
                    }
                }

                create = false;
            }
            catch (ServiceBusException e) when (e.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
            {
            }

            if (create)
            {
                var created = false;
                try
                {
                    LogContext.Debug?.Log("Creating subscription {Subscription} {Topic} -> {ForwardTo}", createSubscriptionOptions.SubscriptionName,
                        createSubscriptionOptions.TopicName,
                        createSubscriptionOptions.ForwardTo);

                    subscriptionProperties = rule != null
                        ? await CreateSubscriptionAsync(createSubscriptionOptions, rule).ConfigureAwait(false)
                        : filter != null
                            ? await CreateSubscriptionAsync(createSubscriptionOptions, filter).ConfigureAwait(false)<commit_msg>Fix potential null reference<commit_after>﻿namespace MassTransit.AzureServiceBusTransport
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;
    using Azure.Messaging.ServiceBus;
    using Azure.Messaging.ServiceBus.Administration;
    using Internals;
    using MassTransit.Middleware;
    using Transports;


    public class ServiceBusConnectionContext :
        BasePipeContext,
        ConnectionContext,
        IAsyncDisposable
    {
        readonly ServiceBusAdministrationClient _administrationClient;
        readonly ServiceBusClient _client;

        public ServiceBusConnectionContext(ServiceBusClient client, ServiceBusAdministrationClient administrationClient, CancellationToken cancellationToken)
            : base(cancellationToken)
        {
            _client = client;
            _administrationClient = administrationClient;
            Endpoint = new Uri($"sb://{_client.FullyQualifiedNamespace}");
        }

        public Uri Endpoint { get; }

        public ServiceBusProcessor CreateQueueProcessor(ReceiveSettings settings)
        {
            return _client.CreateProcessor(settings.Path, GetProcessorOptions(settings));
        }

        public ServiceBusSessionProcessor CreateQueueSessionProcessor(ReceiveSettings settings)
        {
            return _client.CreateSessionProcessor(settings.Path, GetSessionProcessorOptions(settings));
        }

        public ServiceBusProcessor CreateSubscriptionProcessor(SubscriptionSettings settings)
        {
            return _client.CreateProcessor(settings.CreateTopicOptions.Name, settings.CreateSubscriptionOptions.SubscriptionName,
                GetProcessorOptions(settings));
        }

        public ServiceBusSessionProcessor CreateSubscriptionSessionProcessor(SubscriptionSettings settings)
        {
            return _client.CreateSessionProcessor(settings.CreateTopicOptions.Name, settings.CreateSubscriptionOptions.SubscriptionName,
                GetSessionProcessorOptions(settings));
        }

        public ServiceBusSender CreateMessageSender(string entityPath)
        {
            return _client.CreateSender(entityPath);
        }

        public async Task<QueueProperties> CreateQueue(CreateQueueOptions createQueueOptions)
        {
            QueueProperties queueProperties = null;
            try
            {
                queueProperties = await GetQueueAsync(createQueueOptions.Name).ConfigureAwait(false);
            }
            catch (ServiceBusException exception) when (exception.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
            {
            }

            if (queueProperties == null)
            {
                try
                {
                    TransportLogMessages.CreateQueue(createQueueOptions.Name);

                    queueProperties = await CreateQueueAsync(createQueueOptions).ConfigureAwait(false);
                }
                catch (ServiceBusException sbe) when (sbe.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
                {
                    queueProperties = await GetQueueAsync(createQueueOptions.Name).ConfigureAwait(false);
                }
            }

            LogContext.Debug?.Log("Queue: {Queue} ({Attributes})", createQueueOptions.Name,
                string.Join(", ",
                    new[]
                    {
                        createQueueOptions.RequiresDuplicateDetection ? "dupe detect" : "",
                        createQueueOptions.DeadLetteringOnMessageExpiration ? "dead letter" : "",
                        createQueueOptions.RequiresSession ? "session" : "",
                        createQueueOptions.AutoDeleteOnIdle != Defaults.AutoDeleteOnIdle
                            ? $"auto-delete: {createQueueOptions.AutoDeleteOnIdle.ToFriendlyString()}"
                            : ""
                    }.Where(x => !string.IsNullOrWhiteSpace(x))));

            return queueProperties;
        }

        public async Task<TopicProperties> CreateTopic(CreateTopicOptions createTopicOptions)
        {
            TopicProperties topicProperties = null;
            try
            {
                topicProperties = await GetTopicAsync(createTopicOptions.Name).ConfigureAwait(false);
            }
            catch (ServiceBusException exception) when (exception.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
            {
            }

            if (topicProperties == null)
            {
                try
                {
                    TransportLogMessages.CreateTopic(createTopicOptions.Name);

                    topicProperties = await CreateTopicAsync(createTopicOptions).ConfigureAwait(false);
                }
                catch (ServiceBusException e) when (e.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists)
                {
                    topicProperties = await GetTopicAsync(createTopicOptions.Name).ConfigureAwait(false);
                }
            }

            LogContext.Debug?.Log("Topic: {Topic} ({Attributes})", createTopicOptions.Name,
                string.Join(", ",
                    new[]
                    {
                        createTopicOptions.RequiresDuplicateDetection ? "dupe detect" : "",
                        createTopicOptions.EnablePartitioning ? "partitioned" : "",
                        createTopicOptions.SupportOrdering ? "ordered" : ""
                    }.Where(x => !string.IsNullOrWhiteSpace(x))));

            return topicProperties;
        }

        public async Task<SubscriptionProperties> CreateTopicSubscription(CreateSubscriptionOptions createSubscriptionOptions, CreateRuleOptions rule,
            RuleFilter filter)
        {
            var create = true;
            SubscriptionProperties subscriptionProperties = null;
            try
            {
                subscriptionProperties = await GetSubscriptionAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName)
                    .ConfigureAwait(false);

                string NormalizeForwardTo(string forwardTo)
                {
                    return string.IsNullOrEmpty(forwardTo)
                        ? string.Empty
                        : forwardTo.Replace(Endpoint.ToString(), string.Empty).Trim('/');
                }

                var targetForwardTo = NormalizeForwardTo(createSubscriptionOptions.ForwardTo);
                var currentForwardTo = NormalizeForwardTo(subscriptionProperties.ForwardTo);

                if (!targetForwardTo.Equals(currentForwardTo))
                {
                    LogContext.Debug?.Log("Updating subscription: {Subscription} ({Topic} -> {ForwardTo})", subscriptionProperties.SubscriptionName,
                        subscriptionProperties.TopicName, subscriptionProperties.ForwardTo);

                    await UpdateSubscriptionAsync(subscriptionProperties).ConfigureAwait(false);
                }

                if (rule != null)
                {
                    var ruleProperties = await GetRuleAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName, rule.Name)
                        .ConfigureAwait(false);
                    if (rule.Name == ruleProperties.Name && (!(rule.Filter?.Equals(ruleProperties.Filter) ?? ruleProperties.Filter == null)
                            || !(rule.Action?.Equals(ruleProperties.Action) ?? ruleProperties.Action == null)))
                    {
                        LogContext.Debug?.Log("Updating subscription Rule: {Rule} ({DescriptionFilter} -> {Filter})", rule.Name,
                            ruleProperties.Filter?.ToString(), rule.Filter?.ToString());

                        ruleProperties.Filter = rule.Filter;
                        ruleProperties.Action = rule.Action;

                        await UpdateRuleAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName, ruleProperties)
                            .ConfigureAwait(false);
                    }
                }
                else if (filter != null)
                {
                    IList<RuleProperties> rules = await GetRulesAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName)
                        .ConfigureAwait(false);
                    if (rules.Count == 1)
                    {
                        var existingRule = rules[0];

                        if (Guid.TryParse(existingRule.Name, out _) && !(existingRule.Filter?.Equals(filter) ?? filter == null))
                        {
                            LogContext.Debug?.Log("Updating subscription filter: {Rule} ({DescriptionFilter} -> {Filter})", existingRule.Name,
                                existingRule.Filter?.ToString(), filter?.ToString());

                            existingRule.Filter = filter;

                            await UpdateRuleAsync(createSubscriptionOptions.TopicName, createSubscriptionOptions.SubscriptionName, existingRule)
                                .ConfigureAwait(false);
                        }
                    }
                }

                create = false;
            }
            catch (ServiceBusException e) when (e.Reason == ServiceBusFailureReason.MessagingEntityNotFound)
            {
            }

            if (create)
            {
                var created = false;
                try
                {
                    LogContext.Debug?.Log("Creating subscription {Subscription} {Topic} -> {ForwardTo}", createSubscriptionOptions.SubscriptionName,
                        createSubscriptionOptions.TopicName,
                        createSubscriptionOptions.ForwardTo);

                    subscriptionProperties = rule != null
                        ? await CreateSubscriptionAsync(createSubscriptionOptions, rule).ConfigureAwait(false)
                        : filter != null
                            ? await CreateSubscriptionAsync(createSubscriptionOptions, filter).ConfigureAwait(false)
{'fraction_non_alphanumeric': 0.07499254547261704, 'fraction_numerical': 0.00019878739687903787, 'mean_word_length': 1.8714326484018264, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 23, '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': '3879041', 'n_tokens_mistral': 4352, 'n_tokens_neox': 4176, 'n_words': 702}