<commit_before>﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Foundatio.Extensions;
using Foundatio.Logging;
using Foundatio.Serializer;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using Nito.AsyncEx;

namespace Foundatio.Queues {
    public class AzureServiceBusQueue<T> : QueueBase<T> where T : class {
        private readonly string _connectionString;
        private readonly string _queueName;
        private readonly NamespaceManager _namespaceManager;
        private QueueClient _queueClient;
        private long _enqueuedCount;
        private long _dequeuedCount;
        private long _completedCount;
        private long _abandonedCount;
        private long _workerErrorCount;
        private readonly int _retries;
        private readonly RetryPolicy _retryPolicy;
        private readonly TimeSpan _workItemTimeout = TimeSpan.FromMinutes(5);
        private readonly TimeSpan _autoDeleteOnIdle = TimeSpan.MaxValue;
        private readonly TimeSpan _defaultMessageTimeToLive = TimeSpan.MaxValue;
        private readonly AsyncLock _lock = new AsyncLock();

        public AzureServiceBusQueue(string connectionString, string queueName = null, int retries = 2, TimeSpan? workItemTimeout = null, RetryPolicy retryPolicy = null, ISerializer serializer = null, IEnumerable<IQueueBehavior<T>> behaviors = null, ILoggerFactory loggerFactory = null, TimeSpan? autoDeleteOnIdle = null, TimeSpan? defaultMessageTimeToLive = null) : base(serializer, behaviors, loggerFactory) {
            _connectionString = connectionString;
            _queueName = queueName ?? typeof(T).Name;
            _namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
            _retries = retries;
            _retryPolicy = retryPolicy;

            if (workItemTimeout.HasValue && workItemTimeout.Value < TimeSpan.FromMinutes(5)) {
                _workItemTimeout = workItemTimeout.Value;
            }

            if (autoDeleteOnIdle.HasValue && autoDeleteOnIdle.Value >= TimeSpan.FromMinutes(5)) {
                _autoDeleteOnIdle = autoDeleteOnIdle.Value;
            }

            if (defaultMessageTimeToLive.HasValue && defaultMessageTimeToLive.Value > TimeSpan.Zero) {
                _defaultMessageTimeToLive = defaultMessageTimeToLive.Value;
            }
        }
            
        protected override async Task EnsureQueueCreatedAsync(CancellationToken cancellationToken = new CancellationToken()) {
            if (_queueClient != null) {
                return;
            }

            using (await _lock.LockAsync(cancellationToken)) {
                if (_queueClient != null) {
                    return;
                }

                QueueDescription queueDescription;

                if (!await _namespaceManager.QueueExistsAsync(_queueName).AnyContext()) {
                    try {
                        queueDescription = await _namespaceManager.CreateQueueAsync(new QueueDescription(_queueName) {
                            MaxDeliveryCount = _retries + 1,
                            LockDuration = _workItemTimeout,
                            AutoDeleteOnIdle = _autoDeleteOnIdle,
                            DefaultMessageTimeToLive = _defaultMessageTimeToLive
                        }).AnyContext();
                    }
                    catch (MessagingEntityAlreadyExistsException) {
                        queueDescription = await _namespaceManager.GetQueueAsync(_queueName).AnyContext();
                    }
                }
                else {
                    queueDescription = await _namespaceManager.GetQueueAsync(_queueName).AnyContext();

                    bool changes = false;

                    int newMaxDeliveryCount = _retries + 1;
                    if (queueDescription.MaxDeliveryCount != newMaxDeliveryCount) {
                        queueDescription.MaxDeliveryCount = newMaxDeliveryCount;
                        changes = true;
                    }

                    if (queueDescription.LockDuration != _workItemTimeout) {
                        queueDescription.LockDuration = _workItemTimeout;
                        changes = true;
                    }

                    if (queueDescription.AutoDeleteOnIdle != _autoDeleteOnIdle) {
                        queueDescription.AutoDeleteOnIdle = _autoDeleteOnIdle;
                        changes = true;
                    }

                    if (queueDescription.DefaultMessageTimeToLive != _defaultMessageTimeToLive) {
                        queueDescription.DefaultMessageTimeToLive = _defaultMessageTimeToLive;
                        changes = true;
                    }

                    if (changes) {<commit_msg>need to catch base exception, or certain 409 errors don't get caught and retried<commit_after>﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Foundatio.Extensions;
using Foundatio.Logging;
using Foundatio.Serializer;
using Microsoft.ServiceBus;
using Microsoft.ServiceBus.Messaging;
using Nito.AsyncEx;

namespace Foundatio.Queues {
    public class AzureServiceBusQueue<T> : QueueBase<T> where T : class {
        private readonly string _connectionString;
        private readonly string _queueName;
        private readonly NamespaceManager _namespaceManager;
        private QueueClient _queueClient;
        private long _enqueuedCount;
        private long _dequeuedCount;
        private long _completedCount;
        private long _abandonedCount;
        private long _workerErrorCount;
        private readonly int _retries;
        private readonly RetryPolicy _retryPolicy;
        private readonly TimeSpan _workItemTimeout = TimeSpan.FromMinutes(5);
        private readonly TimeSpan _autoDeleteOnIdle = TimeSpan.MaxValue;
        private readonly TimeSpan _defaultMessageTimeToLive = TimeSpan.MaxValue;
        private readonly AsyncLock _lock = new AsyncLock();

        public AzureServiceBusQueue(string connectionString, string queueName = null, int retries = 2, TimeSpan? workItemTimeout = null, RetryPolicy retryPolicy = null, ISerializer serializer = null, IEnumerable<IQueueBehavior<T>> behaviors = null, ILoggerFactory loggerFactory = null, TimeSpan? autoDeleteOnIdle = null, TimeSpan? defaultMessageTimeToLive = null) : base(serializer, behaviors, loggerFactory) {
            _connectionString = connectionString;
            _queueName = queueName ?? typeof(T).Name;
            _namespaceManager = NamespaceManager.CreateFromConnectionString(connectionString);
            _retries = retries;
            _retryPolicy = retryPolicy;

            if (workItemTimeout.HasValue && workItemTimeout.Value < TimeSpan.FromMinutes(5)) {
                _workItemTimeout = workItemTimeout.Value;
            }

            if (autoDeleteOnIdle.HasValue && autoDeleteOnIdle.Value >= TimeSpan.FromMinutes(5)) {
                _autoDeleteOnIdle = autoDeleteOnIdle.Value;
            }

            if (defaultMessageTimeToLive.HasValue && defaultMessageTimeToLive.Value > TimeSpan.Zero) {
                _defaultMessageTimeToLive = defaultMessageTimeToLive.Value;
            }
        }
            
        protected override async Task EnsureQueueCreatedAsync(CancellationToken cancellationToken = new CancellationToken()) {
            if (_queueClient != null) {
                return;
            }

            using (await _lock.LockAsync(cancellationToken)) {
                if (_queueClient != null) {
                    return;
                }

                QueueDescription queueDescription;

                if (!await _namespaceManager.QueueExistsAsync(_queueName).AnyContext()) {
                    try {
                        queueDescription = await _namespaceManager.CreateQueueAsync(new QueueDescription(_queueName) {
                            MaxDeliveryCount = _retries + 1,
                            LockDuration = _workItemTimeout,
                            AutoDeleteOnIdle = _autoDeleteOnIdle,
                            DefaultMessageTimeToLive = _defaultMessageTimeToLive
                        }).AnyContext();
                    }
                    catch (MessagingException) {
                        queueDescription = await _namespaceManager.GetQueueAsync(_queueName).AnyContext();
                    }
                }
                else {
                    queueDescription = await _namespaceManager.GetQueueAsync(_queueName).AnyContext();

                    bool changes = false;

                    int newMaxDeliveryCount = _retries + 1;
                    if (queueDescription.MaxDeliveryCount != newMaxDeliveryCount) {
                        queueDescription.MaxDeliveryCount = newMaxDeliveryCount;
                        changes = true;
                    }

                    if (queueDescription.LockDuration != _workItemTimeout) {
                        queueDescription.LockDuration = _workItemTimeout;
                        changes = true;
                    }

                    if (queueDescription.AutoDeleteOnIdle != _autoDeleteOnIdle) {
                        queueDescription.AutoDeleteOnIdle = _autoDeleteOnIdle;
                        changes = true;
                    }

                    if (queueDescription.DefaultMessageTimeToLive != _defaultMessageTimeToLive) {
                        queueDescription.DefaultMessageTimeToLive = _defaultMessageTimeToLive;
                        changes = true;
                    }

                    if (changes) {
{'fraction_non_alphanumeric': 0.061254152823920265, 'fraction_numerical': 0.0015573089700996678, 'mean_word_length': 2.0159674389480275, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, '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': '6600547', 'n_tokens_mistral': 2249, 'n_tokens_neox': 2157, 'n_words': 483}