QueueTestBase.cs

using Exceptionless;
using Foundatio.AsyncEx;
using Foundatio.Caching;
using Foundatio.Jobs;
using Foundatio.Lock;
using Foundatio.Xunit;
using Foundatio.Messaging;
using Foundatio.Metrics;
using Foundatio.Queues;
using Foundatio.Tests.Extensions;
using Foundatio.Utility;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Xunit;
using Xunit.Abstractions;
using Foundatio.Tests.Metrics;
#pragma warning disable CS4014

namespace Foundatio.Tests.Queue {
    public abstract class QueueTestBase : TestWithLoggingBase, IDisposable {
        protected QueueTestBase(ITestOutputHelper output) : base(output) {
            Log.SetLogLevel<InMemoryCacheClient>(LogLevel.Debug);
            Log.SetLogLevel<InMemoryMetricsClient>(LogLevel.Debug);
            Log.SetLogLevel<MetricsQueueBehavior<SimpleWorkItem>>(LogLevel.Debug);
            Log.SetLogLevel<ScheduledTimer>(LogLevel.Debug);
        }

        protected virtual IQueue<SimpleWorkItem> GetQueue(int retries = 1, TimeSpan? workItemTimeout = null, TimeSpan? retryDelay = null, int[] retryMultipliers = null, int deadLetterMaxItems = 100, bool runQueueMaintenance = true) {
            return null;
        }

        protected virtual async Task CleanupQueueAsync(IQueue<SimpleWorkItem> queue) {
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
            } catch (Exception ex) {
                if (_logger.IsEnabled(LogLevel.Error))
                    _logger.LogError(ex, "Error cleaning up queue");
            } finally {
                queue.Dispose();
            }
        }
        
        protected bool _assertStats = true;

        public virtual async Task CanQueueAndDequeueWorkItemAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello",
                    SubMetricName = "myitem"
                });
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                if (_assertStats)
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);

                await workItem.CompleteAsync();
                Assert.False(workItem.IsAbandoned);
                Assert.True(workItem.IsCompleted);

                metricsCollector.RecordObservableInstruments();
                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Completed);
                    Assert.Equal(0, stats.Queued);

                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.dequeued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.completed"));

                    Assert.Equal(0, metricsCollector.GetSum<int>("foundatio.simpleworkitem.count"));
                    Assert.Equal(0, metricsCollector.GetSum<int>("foundatio.simpleworkitem.working"));
                    Assert.Equal(0, metricsCollector.GetSum<int>("foundatio.simpleworkitem.deadletter"));

                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.myitem.enqueued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.myitem.dequeued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.myitem.completed"));
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanQueueAndDequeueWorkItemWithDelayAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                }, new QueueEntryOptions { DeliveryDelay = TimeSpan.FromSeconds(1) });
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.Null(workItem);

                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(2));
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                if (_assertStats)
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);

                await workItem.CompleteAsync();
                Assert.False(workItem.IsAbandoned);
                Assert.True(workItem.IsCompleted);
                
                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Completed);
                    Assert.Equal(0, stats.Queued);

                    metricsCollector.RecordObservableInstruments();
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.dequeued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.completed"));

                    Assert.Equal(0, metricsCollector.GetSum<long>("foundatio.simpleworkitem.count"));
                    Assert.Equal(0, metricsCollector.GetSum<long>("foundatio.simpleworkitem.working"));
                    Assert.Equal(0, metricsCollector.GetSum<long>("foundatio.simpleworkitem.deadletter"));
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanUseQueueOptionsAsync() {
            var queue = GetQueue(retryDelay: TimeSpan.Zero);
            if (queue == null)
                return;

            using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

            try {
                using var listener = new ActivityListener {
                    ShouldListenTo = s => s.Name == "Foundatio",
                    Sample = (ref ActivityCreationOptions<ActivityContext> _) => ActivitySamplingResult.AllDataAndRecorded,
                    ActivityStarted = activity => _logger.LogInformation("Start: " + activity.DisplayName),
                    ActivityStopped = activity => _logger.LogInformation("Stop: " + activity.DisplayName)
                };

                Activity.Current = new Activity("Parent");

                ActivitySource.AddActivityListener(listener);

                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                }, new QueueEntryOptions {
                    CorrelationId = "123+456",
                    Properties = new Dictionary<string, string> {
                        { "hey", "now" }
                    }
                });
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal("123+456", workItem.CorrelationId);
                Assert.Single(workItem.Properties);
                Assert.Contains(workItem.Properties, i => i.Key == "hey" && i.Value.ToString() == "now");
                if (_assertStats) {
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.dequeued"));
                }

                await workItem.AbandonAsync();
                Assert.True(workItem.IsAbandoned);
                Assert.False(workItem.IsCompleted);
                await Task.Delay(100);

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Abandoned);
                    Assert.Equal(0, stats.Completed);
                    Assert.Equal(1, stats.Queued);

                    metricsCollector.RecordObservableInstruments();
                    Assert.Equal(0, metricsCollector.GetSum<int>("foundatio.simpleworkitem.completed"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.abandoned"));
                    Assert.Equal(1, metricsCollector.GetSum<long>("foundatio.simpleworkitem.count"));
                }

                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(10));
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal("123+456", workItem.CorrelationId);
                Assert.Equal(2, workItem.Attempts);
                Assert.Single(workItem.Properties);
                Assert.Contains(workItem.Properties, i => i.Key == "hey" && i.Value.ToString() == "now");
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanDiscardDuplicateQueueEntriesAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);
                queue.AttachBehavior(new DuplicateDetectionQueueBehavior<SimpleWorkItem>(new InMemoryCacheClient(), Log));
                
                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello",
                    UniqueIdentifier = "123"
                });
                if (_assertStats) {
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                }

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello",
                    UniqueIdentifier = "123"
                });
                if (_assertStats) {
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                }

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                if (_assertStats) {
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.dequeued"));
                }

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello",
                    UniqueIdentifier = "123"
                });
                if (_assertStats) {
                    Assert.Equal(2, (await queue.GetQueueStatsAsync()).Enqueued);
                    Assert.Equal(2, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                }

                await workItem.CompleteAsync();
                Assert.False(workItem.IsAbandoned);
                Assert.True(workItem.IsCompleted);
                var stats = await queue.GetQueueStatsAsync();
                if (_assertStats) {
                    Assert.Equal(1, stats.Completed);
                    Assert.Equal(1, stats.Queued);

                    metricsCollector.RecordObservableInstruments();
                    Assert.Equal(2, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.dequeued"));
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.completed"));
                    Assert.Equal(0, metricsCollector.GetSum<int>("foundatio.simpleworkitem.abandoned"));

                    Assert.Equal(1, metricsCollector.GetSum<long>("foundatio.simpleworkitem.count"));
                    Assert.Equal(0, metricsCollector.GetSum<long>("foundatio.simpleworkitem.working"));
                    Assert.Equal(0, metricsCollector.GetSum<long>("foundatio.simpleworkitem.deadletter"));
                }

            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual Task VerifyRetryAttemptsAsync() {
            const int retryCount = 2;
            var queue = GetQueue(retryCount, TimeSpan.FromSeconds(1), TimeSpan.Zero, new []{ 1 });
            if (queue == null)
                return Task.CompletedTask;

            return VerifyRetryAttemptsImplAsync(queue, retryCount, TimeSpan.FromSeconds(10));
        }

        public virtual Task VerifyDelayedRetryAttemptsAsync() {
            const int retryCount = 2;
            var queue = GetQueue(retryCount, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(1), new []{ 1 });
            if (queue == null)
                return Task.CompletedTask;

            return VerifyRetryAttemptsImplAsync(queue, retryCount, TimeSpan.FromSeconds(7));
        }

        private async Task VerifyRetryAttemptsImplAsync(IQueue<SimpleWorkItem> queue, int retryCount, TimeSpan waitTime) {
            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

                var countdown = new AsyncCountdownEvent(retryCount + 1);
                int attempts = 0;
                await queue.StartWorkingAsync(async w => {
                    Interlocked.Increment(ref attempts);
                    _logger.LogInformation("Starting Attempt {Attempt} to work on queue item", attempts);
                    Assert.Equal("Hello", w.Value.Data);

                    var queueEntryMetadata = (IQueueEntryMetadata) w;
                    Assert.Equal(attempts, queueEntryMetadata.Attempts);

                    await w.AbandonAsync();
                    countdown.Signal();
                    
                    _logger.LogInformation("Finished Attempt {Attempt} to work on queue item, Metadata Attempts: {MetadataAttempts}", attempts, queueEntryMetadata.Attempts);
                });

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });

                await countdown.WaitAsync(waitTime);
                Assert.Equal(0, countdown.CurrentCount);

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(retryCount + 1, attempts);
                    Assert.Equal(0, stats.Completed);
                    Assert.Equal(0, stats.Queued);
                    Assert.Equal(0, stats.Errors);
                    Assert.Equal(retryCount + 1, stats.Dequeued);
                    Assert.Equal(retryCount + 1, stats.Abandoned);

                    metricsCollector.RecordObservableInstruments();
                    Assert.Equal(retryCount + 1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.dequeued"));
                    Assert.Equal(0, metricsCollector.GetSum<int>("foundatio.simpleworkitem.completed"));
                    Assert.Equal(retryCount + 1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.abandoned"));

                    Assert.Equal(0, metricsCollector.GetSum<long>("foundatio.simpleworkitem.count"));
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        /// <summary>
        /// When a cancelled token is passed into Dequeue, it will only try to dequeue one time and then exit.
        /// </summary>
        /// <returns></returns>
        public virtual async Task CanDequeueWithCancelledTokenAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });
                if (_assertStats) {
                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.enqueued"));
                }

                var workItem = await queue.DequeueAsync(new CancellationToken(true));
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);

                // TODO: We should verify that only one retry occurred.
                await workItem.CompleteAsync();

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Completed);
                    Assert.Equal(0, stats.Queued);
                    Assert.Equal(1, metricsCollector.GetSum<int>("foundatio.simpleworkitem.completed"));
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanDequeueEfficientlyAsync() {
            const int iterations = 100;

            var queue = GetQueue(runQueueMaintenance: false);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);
                await queue.EnqueueAsync(new SimpleWorkItem { Data = "Initialize queue to create more accurate metrics" });
                Assert.NotNull(await queue.DequeueAsync(TimeSpan.FromSeconds(1)));

                using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

                using var metrics = new InMemoryMetricsClient(new InMemoryMetricsClientOptions());
                queue.AttachBehavior(new MetricsQueueBehavior<SimpleWorkItem>(metrics, reportCountsInterval: TimeSpan.FromMilliseconds(100), loggerFactory: Log));

                _ = Task.Run(async () => {
                    _logger.LogTrace("Starting enqueue loop");
                    for (int index = 0; index < iterations; index++) {
                        await SystemClock.SleepAsync(RandomData.GetInt(10, 30));
                        await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });
                    }
                    _logger.LogTrace("Finished enqueuing");
                });

                _logger.LogTrace("Starting dequeue loop");
                for (int index = 0; index < iterations; index++) {
                    var item = await queue.DequeueAsync(TimeSpan.FromSeconds(3));
                    Assert.NotNull(item);
                    await item.CompleteAsync();
                }
                _logger.LogTrace("Finished dequeuing");

                await metrics.FlushAsync();
                var timing = await metrics.GetTimerStatsAsync("foundatio.simpleworkitem.queuetime");
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("AverageDuration: {AverageDuration}", timing.AverageDuration);
                Assert.InRange(timing.AverageDuration, 0, 75);
                Assert.InRange(metricsCollector.GetAvg<double>("foundatio.simpleworkitem.queuetime"), 0, 75);
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanResumeDequeueEfficientlyAsync() {
            const int iterations = 10;

            var queue = GetQueue(runQueueMaintenance: false);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                using var metrics = new InMemoryMetricsClient(new InMemoryMetricsClientOptions());
                
                for (int index = 0; index < iterations; index++)
                    await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });

                using var secondQueue = GetQueue(runQueueMaintenance: false);
                secondQueue.AttachBehavior(new MetricsQueueBehavior<SimpleWorkItem>(metrics, reportCountsInterval: TimeSpan.FromMilliseconds(100), loggerFactory: Log));

                _logger.LogTrace("Starting dequeue loop");
                for (int index = 0; index < iterations; index++) {
                    if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("[{Index}] Calling Dequeue", index);
                    var item = await secondQueue.DequeueAsync(TimeSpan.FromSeconds(3));
                    Assert.NotNull(item);
                    await item.CompleteAsync();
                }

                await metrics.FlushAsync(); // This won't flush metrics queue behaviors
                var timing = await metrics.GetTimerStatsAsync("simpleworkitem.queuetime");
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("TotalDuration: {TotalDuration} AverageDuration: {AverageDuration}", timing.TotalDuration, timing.AverageDuration);
                Assert.InRange(timing.AverageDuration, 0, 75);
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanQueueAndDequeueMultipleWorkItemsAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                using var metricsCollector = new DiagnosticsMetricsCollector(FoundatioDiagnostics.Meter.Name, _logger);

                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                const int workItemCount = 25;
                for (int i = 0; i < workItemCount; i++) {
                    await queue.EnqueueAsync(new SimpleWorkItem {
                        Data = "Hello"
                    });
                }
                metricsCollector.RecordObservableInstruments();
                Assert.Equal(workItemCount, metricsCollector.GetMax<long>("foundatio.simpleworkitem.count"));
                Assert.Equal(workItemCount, (await queue.GetQueueStatsAsync()).Queued);

                var sw = Stopwatch.StartNew();
                for (int i = 0; i < workItemCount; i++) {
                    var workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
                    Assert.NotNull(workItem);
                    Assert.Equal("Hello", workItem.Value.Data);
                    await workItem.CompleteAsync();
                }
                sw.Stop();
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                Assert.InRange(sw.Elapsed.TotalSeconds, 0, 5);

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(workItemCount, stats.Dequeued);
                    Assert.Equal(workItemCount, stats.Completed);
                    Assert.Equal(0, stats.Queued);
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task WillNotWaitForItemAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                var sw = Stopwatch.StartNew();
                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                sw.Stop();
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                Assert.Null(workItem);
                Assert.InRange(sw.Elapsed.TotalMilliseconds, 0, 100);
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task WillWaitForItemAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                Log.MinimumLevel = LogLevel.Trace;
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                var sw = Stopwatch.StartNew();
                var workItem = await queue.DequeueAsync(TimeSpan.FromMilliseconds(100));
                sw.Stop();
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                Assert.Null(workItem);
                Assert.InRange(sw.Elapsed, TimeSpan.FromMilliseconds(50), TimeSpan.FromMilliseconds(5000));

                _ = Task.Run(async () => {
                    await Task.Delay(500);
                    _logger.LogInformation("Enqueuing async message");
                    await queue.EnqueueAsync(new SimpleWorkItem {
                        Data = "Hello"
                    });
                });

                sw.Restart();
                _logger.LogInformation("Dequeuing message with timeout");
                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(10));
                sw.Stop();
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                Assert.True(sw.Elapsed > TimeSpan.FromMilliseconds(400));
                Assert.NotNull(workItem);
                await workItem.CompleteAsync();

            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task DequeueWaitWillGetSignaledAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                _ = Task.Run(async () => {
                    await SystemClock.SleepAsync(250);
                    await queue.EnqueueAsync(new SimpleWorkItem {
                        Data = "Hello"
                    });
                });

                var sw = Stopwatch.StartNew();
                var workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(2));
                sw.Stop();
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                Assert.NotNull(workItem);
                Assert.InRange(sw.Elapsed.TotalSeconds, 0, 2);
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanUseQueueWorkerAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                var resetEvent = new AsyncManualResetEvent(false);
                await queue.StartWorkingAsync(async w => {
                    Assert.Equal("Hello", w.Value.Data);
                    await w.CompleteAsync();
                    resetEvent.Set();
                });

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });

                await resetEvent.WaitAsync();
                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Completed);
                    Assert.Equal(0, stats.Queued);
                    Assert.Equal(0, stats.Errors);
                }
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanHandleErrorInWorkerAsync() {
            var queue = GetQueue(retries: 0);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                using var metrics = new InMemoryMetricsClient(new InMemoryMetricsClientOptions { Buffered = false, LoggerFactory = Log });
                
                queue.AttachBehavior(new MetricsQueueBehavior<SimpleWorkItem>(metrics, reportCountsInterval: TimeSpan.FromMilliseconds(100), loggerFactory: Log));
                await queue.StartWorkingAsync(w => {
                    _logger.LogDebug("WorkAction");
                    Assert.Equal("Hello", w.Value.Data);
                    throw new Exception();
                });

                var resetEvent = new AsyncManualResetEvent(false);
                using (queue.Abandoned.AddSyncHandler((o, args) => resetEvent.Set())) {
                    await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });
                    await resetEvent.WaitAsync(TimeSpan.FromSeconds(200));

                    await SystemClock.SleepAsync(100); // give time for the stats to reflect the changes.

                    if (_assertStats) {
                        var stats = await queue.GetQueueStatsAsync();
                        if (_logger.IsEnabled(LogLevel.Information))
                            _logger.LogInformation("Completed: {Completed} Errors: {Errors} Deadletter: {Deadletter} Working: {Working} ", stats.Completed, stats.Errors, stats.Deadletter, stats.Working);
                        Assert.Equal(0, stats.Completed);
                        Assert.Equal(1, stats.Errors);
                        Assert.Equal(1, stats.Deadletter);
                    }
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task WorkItemsWillTimeoutAsync() {
            Log.MinimumLevel = LogLevel.Trace;
            Log.SetLogLevel("Foundatio.Queues.RedisQueue", LogLevel.Trace);
            var queue = GetQueue(retryDelay: TimeSpan.Zero, workItemTimeout: TimeSpan.FromMilliseconds(50));
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });
                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                
                var sw = Stopwatch.StartNew();
                if (_assertStats) {
                    // wait for the entry to be auto abandoned
                    do {
                        var stats = await queue.GetQueueStatsAsync();
                        if (stats.Abandoned > 0)
                            break;
                        await Task.Delay(250);
                    } while (sw.Elapsed < TimeSpan.FromSeconds(10));
                }

                // should throw because the item has already been auto abandoned
                if (_assertStats)
                    await Assert.ThrowsAnyAsync<Exception>(async () => await workItem.CompleteAsync().AnyContext());

                sw = Stopwatch.StartNew();
                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
                sw.Stop();
                _logger.LogTrace("Time {Elapsed:g}", sw.Elapsed);
                Assert.NotNull(workItem);
                await workItem.CompleteAsync();
                if (_assertStats)
                    Assert.Equal(0, (await queue.GetQueueStatsAsync()).Queued);
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task WorkItemsWillGetMovedToDeadletterAsync() {
            var queue = GetQueue(retryDelay: TimeSpan.Zero);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });
                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);

                await workItem.AbandonAsync();
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Abandoned);

                // work item should be retried 1 time.
                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal(2, (await queue.GetQueueStatsAsync()).Dequeued);

                await workItem.AbandonAsync();

                if (_assertStats) {
                    // work item should be moved to deadletter _queue after retries.
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Deadletter);
                    Assert.Equal(2, stats.Abandoned);
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanAutoCompleteWorkerAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                var resetEvent = new AsyncManualResetEvent(false);
                await queue.StartWorkingAsync(w => {
                    Assert.Equal("Hello", w.Value.Data);
                    return Task.CompletedTask;
                }, true);

                using (queue.Completed.AddSyncHandler((s, e) => { resetEvent.Set(); })) {
                    await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });

                    Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);
                    await resetEvent.WaitAsync(TimeSpan.FromSeconds(2));

                    if (_assertStats) {
                        var stats = await queue.GetQueueStatsAsync();
                        Assert.Equal(0, stats.Queued);
                        Assert.Equal(0, stats.Errors);
                        Assert.Equal(1, stats.Completed);
                    }
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanHaveMultipleQueueInstancesAsync() {
            var queue = GetQueue(retries: 0, retryDelay: TimeSpan.Zero);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                const int workItemCount = 500;
                const int workerCount = 3;
                var countdown = new AsyncCountdownEvent(workItemCount);
                var info = new WorkInfo();
                var workers = new List<IQueue<SimpleWorkItem>> { queue };

                try {
                    for (int i = 0; i < workerCount; i++) {
                        var q = GetQueue(retries: 0, retryDelay: TimeSpan.Zero);
                        if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Queue Id: {Id}, I: {Instance}", q.QueueId, i);
                        await q.StartWorkingAsync(w => DoWorkAsync(w, countdown, info));
                        workers.Add(q);
                    }

                    await Run.InParallelAsync(workItemCount, async i => {
                        string id = await queue.EnqueueAsync(new SimpleWorkItem {
                            Data = "Hello",
                            Id = i
                        });
                        if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Enqueued Index: {Instance} Id: {Id}", i, id);
                    });

                    await countdown.WaitAsync();
                    await SystemClock.SleepAsync(50);
                    
                    if (_logger.IsEnabled(LogLevel.Information))
                        _logger.LogInformation("Work Info Stats: Completed: {Completed} Abandoned: {Abandoned} Error: {Errors}", info.CompletedCount, info.AbandonCount, info.ErrorCount);
                    Assert.Equal(workItemCount, info.CompletedCount + info.AbandonCount + info.ErrorCount);

                    // In memory queue doesn't share state.
                    if (queue.GetType() == typeof(InMemoryQueue<SimpleWorkItem>)) {
                        var stats = await queue.GetQueueStatsAsync();
                        Assert.Equal(0, stats.Working);
                        Assert.Equal(0, stats.Timeouts);
                        Assert.Equal(workItemCount, stats.Enqueued);
                        Assert.Equal(workItemCount, stats.Dequeued);
                        Assert.Equal(info.CompletedCount, stats.Completed);
                        Assert.Equal(info.ErrorCount, stats.Errors);
                        Assert.Equal(info.AbandonCount, stats.Abandoned - info.ErrorCount);
                        Assert.Equal(info.AbandonCount + stats.Errors, stats.Deadletter);
                    } else if (_assertStats) {
                        var workerStats = new List<QueueStats>();
                        for (int i = 0; i < workers.Count; i++) {
                            var stats = await workers[i].GetQueueStatsAsync();
                            if (_logger.IsEnabled(LogLevel.Information))
                                _logger.LogInformation("Worker#{Id} Working: {Working} Completed: {Completed} Abandoned: {Abandoned} Error: {Errors} Deadletter: {Deadletter}", i, stats.Working, stats.Completed, stats.Abandoned, stats.Errors, stats.Deadletter);
                            workerStats.Add(stats);
                        }

                        Assert.Equal(info.CompletedCount, workerStats.Sum(s => s.Completed));
                        Assert.Equal(info.ErrorCount, workerStats.Sum(s => s.Errors));
                        Assert.Equal(info.AbandonCount, workerStats.Sum(s => s.Abandoned) - info.ErrorCount);
                        Assert.Equal(info.AbandonCount + workerStats.Sum(s => s.Errors), (workerStats.LastOrDefault()?.Deadletter ?? 0));
                        //Expected: 260
                        //Actual:   125
                    }
                } finally {
                    foreach (var q in workers)
                        await CleanupQueueAsync(q);
                }
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanDelayRetryAsync() {
            var queue = GetQueue(workItemTimeout: TimeSpan.FromMilliseconds(500), retryDelay: TimeSpan.FromSeconds(1));
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);

                var startTime = SystemClock.UtcNow;
                await workItem.AbandonAsync();
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Abandoned);

                workItem = await queue.DequeueAsync(TimeSpan.FromSeconds(5));
                var elapsed = SystemClock.UtcNow.Subtract(startTime);
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Time {Elapsed}", elapsed);
                Assert.NotNull(workItem);
                Assert.True(elapsed > TimeSpan.FromSeconds(.95));
                await workItem.CompleteAsync();
                
                if (_assertStats)
                    Assert.Equal(0, (await queue.GetQueueStatsAsync()).Queued);
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanRunWorkItemWithMetricsAsync() {
            int completedCount = 0;

            using var metrics = new InMemoryMetricsClient(new InMemoryMetricsClientOptions { Buffered = false, LoggerFactory = Log });
            
            var behavior = new MetricsQueueBehavior<WorkItemData>(metrics, "metric", TimeSpan.FromMilliseconds(100), loggerFactory: Log);
            var options = new InMemoryQueueOptions<WorkItemData> { Behaviors = new[] { behavior }, LoggerFactory = Log };
            using var queue = new InMemoryQueue<WorkItemData>(options);
            
            Task Handler(object sender, CompletedEventArgs<WorkItemData> e) {
                completedCount++;
                return Task.CompletedTask;
            }

            using (queue.Completed.AddHandler(Handler)) {
                _logger.LogTrace("Before enqueue");
                await queue.EnqueueAsync(new SimpleWorkItem { Id = 1, Data = "Testing" });
                await queue.EnqueueAsync(new SimpleWorkItem { Id = 2, Data = "Testing" });
                await queue.EnqueueAsync(new SimpleWorkItem { Id = 3, Data = "Testing" });

                await SystemClock.SleepAsync(100);

                _logger.LogTrace("Before dequeue");
                var item = await queue.DequeueAsync();
                await item.CompleteAsync();

                item = await queue.DequeueAsync();
                await item.CompleteAsync();

                item = await queue.DequeueAsync();
                await item.AbandonAsync();

                _logger.LogTrace("Before asserts");
                Assert.Equal(2, completedCount);

                await SystemClock.SleepAsync(100); // flush metrics queue behaviors
                await metrics.FlushAsync();
                Assert.InRange((await metrics.GetGaugeStatsAsync("metric.workitemdata.count")).Max, 1, 3);
                Assert.InRange((await metrics.GetGaugeStatsAsync("metric.workitemdata.working")).Max, 0, 1);

                Assert.Equal(3, await metrics.GetCounterCountAsync("metric.workitemdata.simple.enqueued"));
                Assert.Equal(3, await metrics.GetCounterCountAsync("metric.workitemdata.enqueued"));

                Assert.Equal(3, await metrics.GetCounterCountAsync("metric.workitemdata.simple.dequeued"));
                Assert.Equal(3, await metrics.GetCounterCountAsync("metric.workitemdata.dequeued"));

                Assert.Equal(2, await metrics.GetCounterCountAsync("metric.workitemdata.simple.completed"));
                Assert.Equal(2, await metrics.GetCounterCountAsync("metric.workitemdata.completed"));

                Assert.Equal(1, await metrics.GetCounterCountAsync("metric.workitemdata.simple.abandoned"));
                Assert.Equal(1, await metrics.GetCounterCountAsync("metric.workitemdata.abandoned"));

                var queueTiming = await metrics.GetTimerStatsAsync("metric.workitemdata.simple.queuetime");
                Assert.Equal(3, queueTiming.Count);
                queueTiming = await metrics.GetTimerStatsAsync("metric.workitemdata.queuetime");
                Assert.Equal(3, queueTiming.Count);

                var processTiming = await metrics.GetTimerStatsAsync("metric.workitemdata.simple.processtime");
                Assert.Equal(3, processTiming.Count);
                processTiming = await metrics.GetTimerStatsAsync("metric.workitemdata.processtime");
                Assert.Equal(3, processTiming.Count);

                if (_assertStats) {
                    var queueStats = await metrics.GetQueueStatsAsync("metric.workitemdata");
                    Assert.Equal(3, queueStats.Enqueued.Count);
                    Assert.Equal(3, queueStats.Dequeued.Count);
                    Assert.Equal(2, queueStats.Completed.Count);
                    Assert.Equal(1, queueStats.Abandoned.Count);
                    Assert.InRange(queueStats.Count.Max, 1, 3);
                    Assert.InRange(queueStats.Working.Max, 0, 1);

                    var subQueueStats = await metrics.GetQueueStatsAsync("metric.workitemdata", "simple");
                    Assert.Equal(3, subQueueStats.Enqueued.Count);
                    Assert.Equal(3, subQueueStats.Dequeued.Count);
                    Assert.Equal(2, subQueueStats.Completed.Count);
                    Assert.Equal(1, subQueueStats.Abandoned.Count);
                }
            }
        }

        public virtual async Task CanRenewLockAsync() {
            Log.SetLogLevel<InMemoryQueue<SimpleWorkItem>>(LogLevel.Trace);

            // Need large value to reproduce this test
            var workItemTimeout = TimeSpan.FromSeconds(1);
            // Slightly shorter than the timeout to ensure we haven't lost the lock
            var renewWait = TimeSpan.FromSeconds(workItemTimeout.TotalSeconds * .25d);

            var queue = GetQueue(retryDelay: TimeSpan.Zero, workItemTimeout: workItemTimeout);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello"
                });
                var entry = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(entry);
                Assert.Equal("Hello", entry.Value.Data);

                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Waiting for {RenewWait:g} before renewing lock", renewWait);
                await SystemClock.SleepAsync(renewWait);
                _logger.LogTrace("Renewing lock");
                await entry.RenewLockAsync();
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Waiting for {RenewWait:g} to see if lock was renewed", renewWait);
                await SystemClock.SleepAsync(renewWait);

                // We shouldn't get another item here if RenewLock works.
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Attempting to dequeue item that shouldn't exist");
                var nullWorkItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.Null(nullWorkItem);
                await entry.CompleteAsync();
                
                if (_assertStats)
                    Assert.Equal(0, (await queue.GetQueueStatsAsync()).Queued);
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanAbandonQueueEntryOnceAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);

                await workItem.AbandonAsync();
                Assert.True(workItem.IsAbandoned);
                Assert.False(workItem.IsCompleted);
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.AbandonAsync());
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.CompleteAsync());
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.CompleteAsync());

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(1, stats.Abandoned);
                    Assert.Equal(0, stats.Completed);
                    Assert.Equal(0, stats.Deadletter);
                    Assert.Equal(1, stats.Dequeued);
                    Assert.Equal(1, stats.Enqueued);
                    Assert.Equal(0, stats.Errors);
                    Assert.InRange(stats.Queued, 0, 1);
                    Assert.Equal(0, stats.Timeouts);
                    Assert.Equal(0, stats.Working);
                }

                if (workItem is QueueEntry<SimpleWorkItem> queueEntry)
                    Assert.Equal(1, queueEntry.Attempts);

                await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });
                workItem = await queue.DequeueAsync(TimeSpan.Zero);

                await queue.AbandonAsync(workItem);
                Assert.True(workItem.IsAbandoned);
                Assert.False(workItem.IsCompleted);
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.CompleteAsync());
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.AbandonAsync());
                await Assert.ThrowsAnyAsync<Exception>(() => queue.AbandonAsync(workItem));
                await Assert.ThrowsAnyAsync<Exception>(() => queue.CompleteAsync(workItem));
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanCompleteQueueEntryOnceAsync() {
            var queue = GetQueue();
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });

                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Enqueued);

                var workItem = await queue.DequeueAsync(TimeSpan.Zero);
                Assert.NotNull(workItem);
                Assert.Equal("Hello", workItem.Value.Data);
                Assert.Equal(1, (await queue.GetQueueStatsAsync()).Dequeued);

                await workItem.CompleteAsync();
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.CompleteAsync());
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.AbandonAsync());
                await Assert.ThrowsAnyAsync<Exception>(() => workItem.AbandonAsync());

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(0, stats.Abandoned);
                    Assert.Equal(1, stats.Completed);
                    Assert.Equal(0, stats.Deadletter);
                    Assert.Equal(1, stats.Dequeued);
                    Assert.Equal(1, stats.Enqueued);
                    Assert.Equal(0, stats.Errors);
                    Assert.Equal(0, stats.Queued);
                    Assert.Equal(0, stats.Timeouts);
                    Assert.Equal(0, stats.Working);
                }

                if (workItem is QueueEntry<SimpleWorkItem> queueEntry)
                    Assert.Equal(1, queueEntry.Attempts);
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanDequeueWithLockingAsync() {
            using var cache = new InMemoryCacheClient(new InMemoryCacheClientOptions { LoggerFactory = Log });
            using var messageBus = new InMemoryMessageBus(new InMemoryMessageBusOptions { LoggerFactory = Log });
            
            var distributedLock = new CacheLockProvider(cache, messageBus, Log);
            await CanDequeueWithLockingImpAsync(distributedLock);
        }

        protected async Task CanDequeueWithLockingImpAsync(CacheLockProvider distributedLock) {
            var queue = GetQueue(retryDelay: TimeSpan.Zero, retries: 0);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                Log.MinimumLevel = LogLevel.Trace;
                using var metrics = new InMemoryMetricsClient(new InMemoryMetricsClientOptions { Buffered = false, LoggerFactory = Log });
                
                queue.AttachBehavior(new MetricsQueueBehavior<SimpleWorkItem>(metrics, loggerFactory: Log));

                var resetEvent = new AsyncAutoResetEvent();
                await queue.StartWorkingAsync(async w => {
                    _logger.LogInformation("Acquiring distributed lock in work item");
                    var l = await distributedLock.AcquireAsync("test");
                    Assert.NotNull(l);
                    _logger.LogInformation("Acquired distributed lock");
                    await SystemClock.SleepAsync(TimeSpan.FromMilliseconds(250));
                    await l.ReleaseAsync();
                    _logger.LogInformation("Released distributed lock");

                    await w.CompleteAsync();
                    resetEvent.Set();
                });

                await queue.EnqueueAsync(new SimpleWorkItem { Data = "Hello" });
                await resetEvent.WaitAsync(TimeSpan.FromSeconds(5));

                if (_assertStats) {
                    await SystemClock.SleepAsync(1);
                    var stats = await queue.GetQueueStatsAsync();
                    _logger.LogInformation("Completed: {Completed} Errors: {Errors} Deadletter: {Deadletter} Working: {Working} ", stats.Completed, stats.Errors, stats.Deadletter, stats.Working);
                    Assert.Equal(1, stats.Completed);
                }
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanHaveMultipleQueueInstancesWithLockingAsync() {
            using var cache = new InMemoryCacheClient(new InMemoryCacheClientOptions { LoggerFactory = Log });
            using var messageBus = new InMemoryMessageBus(new InMemoryMessageBusOptions { LoggerFactory = Log });
            
            var distributedLock = new CacheLockProvider(cache, messageBus, Log);
            await CanHaveMultipleQueueInstancesWithLockingImplAsync(distributedLock);
        }

        protected async Task CanHaveMultipleQueueInstancesWithLockingImplAsync(CacheLockProvider distributedLock) {
            var queue = GetQueue(retries: 0, retryDelay: TimeSpan.Zero);
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);

                const int workItemCount = 16;
                const int workerCount = 4;
                var countdown = new AsyncCountdownEvent(workItemCount);
                var info = new WorkInfo();
                var workers = new List<IQueue<SimpleWorkItem>> { queue };

                try {
                    for (int i = 0; i < workerCount; i++) {
                        var q = GetQueue(retries: 0, retryDelay: TimeSpan.Zero);
                        int instanceCount = i;
                        await q.StartWorkingAsync(async w => {
                            if (_logger.IsEnabled(LogLevel.Information))
                                _logger.LogInformation("[{Instance}] Acquiring distributed lock in work item: {Id}", instanceCount, w.Id);
                            var l = await distributedLock.AcquireAsync("test");
                            Assert.NotNull(l);
                            if (_logger.IsEnabled(LogLevel.Information))
                                _logger.LogInformation("[{Instance}] Acquired distributed lock: {Id}", instanceCount, w.Id);
                            await SystemClock.SleepAsync(TimeSpan.FromMilliseconds(50));
                            await l.ReleaseAsync();
                            if (_logger.IsEnabled(LogLevel.Information))
                                _logger.LogInformation("[{Instance}] Released distributed lock: {Id}", instanceCount, w.Id);

                            await w.CompleteAsync();
                            info.IncrementCompletedCount();
                            countdown.Signal();
                            if (_logger.IsEnabled(LogLevel.Information))
                                _logger.LogInformation("[{Instance}] Signaled countdown: {Id}", instanceCount, w.Id);
                        });
                        workers.Add(q);
                    }

                    await Run.InParallelAsync(workItemCount, async i => {
                        string id = await queue.EnqueueAsync(new SimpleWorkItem {
                            Data = "Hello",
                            Id = i
                        });
                        if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Enqueued Index: {Instance} Id: {Id}", i, id);
                    });

                    await countdown.WaitAsync(TimeSpan.FromSeconds(5));
                    await SystemClock.SleepAsync(50);
                    if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Completed: {Completed} Abandoned: {Abandoned} Error: {Errors}", info.CompletedCount, info.AbandonCount, info.ErrorCount);

                    if (_logger.IsEnabled(LogLevel.Information))
                        _logger.LogInformation("Work Info Stats: Completed: {Completed} Abandoned: {Abandoned} Error: {Errors}", info.CompletedCount, info.AbandonCount, info.ErrorCount);
                    Assert.Equal(workItemCount, info.CompletedCount + info.AbandonCount + info.ErrorCount);

                    // In memory queue doesn't share state.
                    if (queue.GetType() == typeof(InMemoryQueue<SimpleWorkItem>)) {
                        var stats = await queue.GetQueueStatsAsync();
                        Assert.Equal(info.CompletedCount, stats.Completed);
                    }
                    else {
                        var workerStats = new List<QueueStats>();
                        for (int i = 0; i < workers.Count; i++) {
                            var stats = await workers[i].GetQueueStatsAsync();
                            if (_logger.IsEnabled(LogLevel.Information))
                                _logger.LogInformation("Worker#{Id} Working: {Working} Completed: {Completed} Abandoned: {Abandoned} Error: {Errors} Deadletter: {Deadletter}", i, stats.Working, stats.Completed, stats.Abandoned, stats.Errors, stats.Deadletter);
                            workerStats.Add(stats);
                        }

                        Assert.Equal(info.CompletedCount, workerStats.Sum(s => s.Completed));
                    }
                }
                finally {
                    foreach (var q in workers)
                        await CleanupQueueAsync(q);
                }
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        protected async Task DoWorkAsync(IQueueEntry<SimpleWorkItem> w, AsyncCountdownEvent countdown, WorkInfo info) {
            if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Starting: {Id}", w.Value.Id);
            Assert.Equal("Hello", w.Value.Data);

            try {
                // randomly complete, abandon or blowup.
                if (RandomData.GetBool()) {
                    if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Completing: {Id}", w.Value.Id);
                    await w.CompleteAsync();
                    info.IncrementCompletedCount();
                }
                else if (RandomData.GetBool()) {
                    if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Abandoning: {Id}", w.Value.Id);
                    await w.AbandonAsync();
                    info.IncrementAbandonCount();
                }
                else {
                    if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Erroring: {Id}", w.Value.Id);
                    info.IncrementErrorCount();
                    throw new Exception();
                }
            }
            finally {
                if (_logger.IsEnabled(LogLevel.Trace)) _logger.LogTrace("Signal {CurrentCount}", countdown.CurrentCount);
                countdown.Signal();
            }
        }

        private async Task AssertEmptyQueueAsync(IQueue<SimpleWorkItem> queue) {
            if (_assertStats) {
                var stats = await queue.GetQueueStatsAsync();
                Assert.Equal(0, stats.Abandoned);
                Assert.Equal(0, stats.Completed);
                Assert.Equal(0, stats.Deadletter);
                Assert.Equal(0, stats.Dequeued);
                Assert.Equal(0, stats.Enqueued);
                Assert.Equal(0, stats.Errors);
                Assert.Equal(0, stats.Queued);
                Assert.Equal(0, stats.Timeouts);
                Assert.Equal(0, stats.Working);
            }
        }

        public virtual async Task MaintainJobNotAbandon_NotWorkTimeOutEntry() {
            var queue = GetQueue(retries: 0, workItemTimeout: TimeSpan.FromMilliseconds(100), retryDelay: TimeSpan.Zero);
            if (queue == null)
                return;
            try {
                await queue.DeleteQueueAsync();
                await AssertEmptyQueueAsync(queue);
                queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello World",
                    Id = 1
                });
                queue.EnqueueAsync(new SimpleWorkItem {
                    Data = "Hello World",
                    Id = 2
                });

                var dequeuedQueueItem = Assert.IsType<QueueEntry<SimpleWorkItem>>(await queue.DequeueAsync());
                Assert.NotNull(dequeuedQueueItem.Value);
                // The first dequeued item works for 60 milliseconds less than work timeout(100 milliseconds).
                await SystemClock.SleepAsync(60);
                await dequeuedQueueItem.CompleteAsync();
                Assert.True(dequeuedQueueItem.IsCompleted);
                Assert.False(dequeuedQueueItem.IsAbandoned);

                dequeuedQueueItem = Assert.IsType<QueueEntry<SimpleWorkItem>>(await queue.DequeueAsync());
                Assert.NotNull(dequeuedQueueItem.Value);
                // The second dequeued item works for 60 milliseconds less than work timeout(100 milliseconds).
                await SystemClock.SleepAsync(60);
                await dequeuedQueueItem.CompleteAsync();
                Assert.True(dequeuedQueueItem.IsCompleted);
                Assert.False(dequeuedQueueItem.IsAbandoned);

                if (_assertStats) {
                    var stats = await queue.GetQueueStatsAsync();
                    Assert.Equal(0, stats.Working);
                    Assert.Equal(0, stats.Abandoned);
                    Assert.Equal(2, stats.Completed);
                }
            }
            finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual async Task CanHandleAutoAbandonInWorker() {
            // create queue with short work item timeout so it will be auto abandoned
            var queue = GetQueue(workItemTimeout: TimeSpan.FromMilliseconds(100));
            if (queue == null)
                return;

            try {
                await queue.DeleteQueueAsync();

                var successEvent = new AsyncAutoResetEvent();
                var errorEvent = new AsyncAutoResetEvent();

                await queue.StartWorkingAsync(async (item) => {
                    if (item.Value.Data == "Delay") {
                        // wait for queue item to get auto abandoned
                        var stats = await queue.GetQueueStatsAsync();
                        var sw = Stopwatch.StartNew();
                        do {
                            if (stats.Abandoned > 0)
                                break;

                            stats = await queue.GetQueueStatsAsync();
                        } while (sw.Elapsed < TimeSpan.FromSeconds(10));

                        Assert.Equal(1, stats.Abandoned);
                    }

                    try {
                        await item.CompleteAsync();
                    } catch {
                        errorEvent.Set();
                        throw;
                    }

                    successEvent.Set();
                });

                await queue.EnqueueAsync(new SimpleWorkItem() { Data = "Delay" });
                await queue.EnqueueAsync(new SimpleWorkItem() { Data = "No Delay" });
                
                await errorEvent.WaitAsync(TimeSpan.FromSeconds(10));
                await successEvent.WaitAsync(TimeSpan.FromSeconds(10));
            } finally {
                await CleanupQueueAsync(queue);
            }
        }

        public virtual void Dispose() {
            var queue = GetQueue();
            if (queue == null)
                return;

            using (queue)
                _ = Task.Run(queue.DeleteQueueAsync);

            GC.SuppressFinalize(this);
        }
    }

    public class WorkInfo {
        private int _abandonCount;
        private int _errorCount;
        private int _completedCount;

        public int AbandonCount => _abandonCount;
        public int ErrorCount => _errorCount;
        public int CompletedCount => _completedCount;

        public void IncrementAbandonCount() {
            Interlocked.Increment(ref _abandonCount);
        }

        public void IncrementErrorCount() {
            Interlocked.Increment(ref _errorCount);
        }

        public void IncrementCompletedCount() {
            Interlocked.Increment(ref _completedCount);
        }
    }
}