From 0919bfee502094e60b254bd14f56714db2864e8b Mon Sep 17 00:00:00 2001 From: Mark van Renswoude Date: Fri, 31 Mar 2023 07:42:59 +0200 Subject: [PATCH] Fixed #41 - RabbitMQ.Client 6.5 compatibility Consumers are now properly async as well --- Tapeti/Connection/TapetiBasicConsumer.cs | 23 ++-- Tapeti/Connection/TapetiChannel.cs | 8 +- Tapeti/Connection/TapetiClient.cs | 16 +-- Tapeti/Tapeti.csproj | 2 +- Tapeti/Tasks/SerialTaskQueue.cs | 65 ++++++++++ Tapeti/Tasks/SingleThreadTaskQueue.cs | 145 ----------------------- 6 files changed, 89 insertions(+), 170 deletions(-) create mode 100644 Tapeti/Tasks/SerialTaskQueue.cs delete mode 100644 Tapeti/Tasks/SingleThreadTaskQueue.cs diff --git a/Tapeti/Connection/TapetiBasicConsumer.cs b/Tapeti/Connection/TapetiBasicConsumer.cs index 5b04ba5..24badec 100644 --- a/Tapeti/Connection/TapetiBasicConsumer.cs +++ b/Tapeti/Connection/TapetiBasicConsumer.cs @@ -17,7 +17,7 @@ namespace Tapeti.Connection /// /// Implements the bridge between the RabbitMQ Client consumer and a Tapeti Consumer /// - internal class TapetiBasicConsumer : DefaultBasicConsumer + internal class TapetiBasicConsumer : AsyncDefaultBasicConsumer { private readonly IConsumer consumer; private readonly long connectionReference; @@ -34,7 +34,7 @@ namespace Tapeti.Connection /// - public override void HandleBasicDeliver(string consumerTag, + public override async Task HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, @@ -50,18 +50,15 @@ namespace Tapeti.Connection // See also: https://github.com/JamesNK/Newtonsoft.Json/issues/1761 var bodyArray = body.ToArray(); - Task.Run(async () => + try { - try - { - var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray); - await onRespond(connectionReference, deliveryTag, response); - } - catch - { - await onRespond(connectionReference, deliveryTag, ConsumeResult.Error); - } - }); + var response = await consumer.Consume(exchange, routingKey, new RabbitMQMessageProperties(properties), bodyArray); + await onRespond(connectionReference, deliveryTag, response); + } + catch + { + await onRespond(connectionReference, deliveryTag, ConsumeResult.Error); + } } } } diff --git a/Tapeti/Connection/TapetiChannel.cs b/Tapeti/Connection/TapetiChannel.cs index ed39446..da75767 100644 --- a/Tapeti/Connection/TapetiChannel.cs +++ b/Tapeti/Connection/TapetiChannel.cs @@ -21,7 +21,7 @@ namespace Tapeti.Connection { private readonly Func modelFactory; private readonly object taskQueueLock = new(); - private SingleThreadTaskQueue? taskQueue; + private SerialTaskQueue? taskQueue; private readonly ModelProvider modelProvider; @@ -34,7 +34,7 @@ namespace Tapeti.Connection public async Task Reset() { - SingleThreadTaskQueue? capturedTaskQueue; + SerialTaskQueue? capturedTaskQueue; lock (taskQueueLock) { @@ -80,11 +80,11 @@ namespace Tapeti.Connection - private SingleThreadTaskQueue GetTaskQueue() + private SerialTaskQueue GetTaskQueue() { lock (taskQueueLock) { - return taskQueue ??= new SingleThreadTaskQueue(); + return taskQueue ??= new SerialTaskQueue(); } } diff --git a/Tapeti/Connection/TapetiClient.cs b/Tapeti/Connection/TapetiClient.cs index 07e2718..04cd078 100644 --- a/Tapeti/Connection/TapetiClient.cs +++ b/Tapeti/Connection/TapetiClient.cs @@ -777,7 +777,8 @@ namespace Tapeti.Connection Password = connectionParams.Password, AutomaticRecoveryEnabled = false, TopologyRecoveryEnabled = false, - RequestedHeartbeat = TimeSpan.FromSeconds(30) + RequestedHeartbeat = TimeSpan.FromSeconds(30), + DispatchConsumersAsync = true }; if (connectionParams.ClientProperties != null) @@ -790,6 +791,9 @@ namespace Tapeti.Connection } + // TODO lock both channels when attempting the connection + // TODO when one channel is lost, do not reconnect, instead restore the channel + while (true) { try @@ -808,25 +812,23 @@ namespace Tapeti.Connection { try { - if (connection is { IsOpen: true }) - connection.Close(); + if (capturedConnection is { IsOpen: true }) + capturedConnection.Close(); } catch (AlreadyClosedException) { } finally { - connection?.Dispose(); + capturedConnection?.Dispose(); } - - connection = null; } logger.Connect(new ConnectContext(connectionParams, isReconnect)); Interlocked.Increment(ref connectionReference); lock (connectionLock) - { + { connection = connectionFactory.CreateConnection(); capturedConnection = connection; diff --git a/Tapeti/Tapeti.csproj b/Tapeti/Tapeti.csproj index 18a03ff..2ba6a69 100644 --- a/Tapeti/Tapeti.csproj +++ b/Tapeti/Tapeti.csproj @@ -25,7 +25,7 @@ - + diff --git a/Tapeti/Tasks/SerialTaskQueue.cs b/Tapeti/Tasks/SerialTaskQueue.cs new file mode 100644 index 0000000..b4d3081 --- /dev/null +++ b/Tapeti/Tasks/SerialTaskQueue.cs @@ -0,0 +1,65 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace Tapeti.Tasks +{ + /// + /// An implementation of a queue which runs tasks serially. + /// + public class SerialTaskQueue : IDisposable + { + private readonly object previousTaskLock = new(); + private Task previousTask = Task.CompletedTask; + + + /// + /// Add the specified synchronous action to the task queue. + /// + /// + public Task Add(Action action) + { + lock (previousTaskLock) + { + previousTask = previousTask.ContinueWith( + _ => action(), + CancellationToken.None, + TaskContinuationOptions.None, + TaskScheduler.Default); + + return previousTask; + } + } + + + /// + /// Add the specified asynchronous method to the task queue. + /// + /// + public Task Add(Func func) + { + lock (previousTaskLock) + { + var task = previousTask.ContinueWith( + _ => func(), + CancellationToken.None, + TaskContinuationOptions.None, + TaskScheduler.Default); + + previousTask = task; + + // 'task' completes at the moment a Task is returned (for example, an await is encountered), + // this is used to chain the next. We return the unwrapped Task however, so that the caller + // awaits until the full task chain has completed. + return task.Unwrap(); + } + } + + + /// + public void Dispose() + { + GC.SuppressFinalize(this); + } + } +} diff --git a/Tapeti/Tasks/SingleThreadTaskQueue.cs b/Tapeti/Tasks/SingleThreadTaskQueue.cs deleted file mode 100644 index 30032bb..0000000 --- a/Tapeti/Tasks/SingleThreadTaskQueue.cs +++ /dev/null @@ -1,145 +0,0 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Threading; -using System.Threading.Tasks; - -namespace Tapeti.Tasks -{ - /// - /// An implementation of a queue which runs tasks on a single thread. - /// - public class SingleThreadTaskQueue : IDisposable - { - private readonly object previousTaskLock = new(); - private Task previousTask = Task.CompletedTask; - - private readonly Lazy singleThreadScheduler = new(); - - - /// - /// Add the specified synchronous action to the task queue. - /// - /// - public Task Add(Action action) - { - lock (previousTaskLock) - { - previousTask = previousTask.ContinueWith(_ => action(), CancellationToken.None - , TaskContinuationOptions.None - , singleThreadScheduler.Value); - - return previousTask; - } - } - - - /// - /// Add the specified asynchronous method to the task queue. - /// - /// - public Task Add(Func func) - { - lock (previousTaskLock) - { - var task = previousTask.ContinueWith(_ => func(), CancellationToken.None - , TaskContinuationOptions.None - , singleThreadScheduler.Value); - - previousTask = task; - - // 'task' completes at the moment a Task is returned (for example, an await is encountered), - // this is used to chain the next. We return the unwrapped Task however, so that the caller - // awaits until the full task chain has completed. - return task.Unwrap(); - } - } - - - /// - public void Dispose() - { - if (singleThreadScheduler.IsValueCreated) - singleThreadScheduler.Value.Dispose(); - } - - - internal class SingleThreadTaskScheduler : TaskScheduler, IDisposable - { - public override int MaximumConcurrencyLevel => 1; - - - private readonly Queue scheduledTasks = new(); - private bool disposed; - - - public SingleThreadTaskScheduler() - { - // ReSharper disable once ObjectCreationAsStatement - fire and forget! - new Thread(WorkerThread).Start(); - } - - - public void Dispose() - { - lock (scheduledTasks) - { - disposed = true; - Monitor.PulseAll(scheduledTasks); - } - } - - - protected override void QueueTask(Task task) - { - if (disposed) return; - - lock (scheduledTasks) - { - scheduledTasks.Enqueue(task); - Monitor.Pulse(scheduledTasks); - } - } - - protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) - { - return false; - } - - - protected override IEnumerable GetScheduledTasks() - { - lock (scheduledTasks) - { - return scheduledTasks.ToList(); - } - } - - - private void WorkerThread() - { - while (true) - { - Task? task; - lock (scheduledTasks) - { - task = WaitAndDequeueTask(); - } - - if (task == null) - break; - - TryExecuteTask(task); - } - } - - private Task? WaitAndDequeueTask() - { - while (!scheduledTasks.Any() && !disposed) - Monitor.Wait(scheduledTasks); - - return disposed ? null : scheduledTasks.Dequeue(); - } - } - } -}