Merge pull request #42 from rabbitmq/rabbitmq-amqp-dotnet-client-37-followup

Implement `UnsettledMessageCount` in `IConsumer`

Author: lukebakken

RabbitMQ.AMQP.Client/IConsumer.cs

@@ -9,9 +9,9 @@ public interface IConsumer : ILifeCycle
 {
     void Pause();
 
-    long UnsettledMessageCount();
-
     void Unpause();
+
+    long UnsettledMessageCount { get; }
 }
 
 public interface IMessageHandler

RabbitMQ.AMQP.Client/IPublisher.cs

@@ -8,20 +8,39 @@ public enum OutcomeState
     Failed,
 }
 
-public class OutcomeDescriptor(ulong code, string description, OutcomeState state, Error? error)
+public class PublishOutcome
 {
-    public OutcomeState State { get; internal set; } = state;
-    public ulong Code { get; internal set; } = code;
-    public string Description { get; internal set; } = description;
+    private readonly OutcomeState _state;
+    private readonly Error? _error;
 
-    public Error? Error { get; internal set; } = error;
+    public PublishOutcome(OutcomeState state, Error? error)
+    {
+        _state = state;
+        _error = error;
+    }
+
+    public OutcomeState State => _state;
+    public Error? Error => _error;
 }
 
-public delegate void OutcomeDescriptorCallback(IMessage message, OutcomeDescriptor outcomeDescriptor);
+// public delegate void OutcomeDescriptorCallback(IMessage message, PublishOutcome publishOutcome);
+
+public class PublishResult
+{
+    private IMessage _message;
+    private PublishOutcome _outcome;
+
+    public PublishResult(IMessage message, PublishOutcome outcome)
+    {
+        _message = message;
+        _outcome = outcome;
+    }
+
+    public IMessage Message => _message;
+    public PublishOutcome Outcome => _outcome;
+}
 
 public interface IPublisher : ILifeCycle
 {
-    // TODO this should be named PublishAsync
-    Task Publish(IMessage message,
-        OutcomeDescriptorCallback outcomeCallback); // TODO: Add CancellationToken and callBack
+    Task<PublishResult> PublishAsync(IMessage message, CancellationToken cancellationToken = default);
 }

RabbitMQ.AMQP.Client/Impl/AmqpConnection.cs

@@ -1,3 +1,7 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+
 using System.Collections.Concurrent;
 using System.Collections.ObjectModel;
 using Amqp;
@@ -343,18 +347,19 @@ await Task.Run(async () =>
                         {
                             try
                             {
-                                int next = _connectionSettings.Recovery.GetBackOffDelayPolicy().Delay();
+                                int nextDelayMs = _connectionSettings.Recovery.GetBackOffDelayPolicy().Delay();
 
                                 Trace.WriteLine(TraceLevel.Information,
-                                    $"Trying Recovering connection in {next} milliseconds, " +
+                                    $"Trying Recovering connection in {nextDelayMs} milliseconds, " +
                                     $"attempt: {_connectionSettings.Recovery.GetBackOffDelayPolicy().CurrentAttempt}. " +
                                     $"Info: {ToString()})");
 
-                                await Task.Delay(TimeSpan.FromMilliseconds(next))
+                                await Task.Delay(TimeSpan.FromMilliseconds(nextDelayMs))
                                     .ConfigureAwait(false);
 
                                 await OpenConnectionAsync()
                                     .ConfigureAwait(false);
+
                                 connected = true;
                             }
                             catch (Exception e)

RabbitMQ.AMQP.Client/Impl/AmqpConsumer.cs

@@ -1,3 +1,7 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2017-2023 Broadcom. All Rights Reserved. The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
+
 using System.Runtime.CompilerServices;
 using Amqp;
 using Amqp.Framing;
@@ -7,20 +11,21 @@ namespace RabbitMQ.AMQP.Client.Impl;
 
 public class AmqpConsumer : AbstractReconnectLifeCycle, IConsumer
 {
-    private enum PauseStatus
+    private enum PauseStatus : byte
     {
-        UNPAUSED,
-        PAUSING,
-        PAUSED
+        UNPAUSED = 0,
+        PAUSING = 1,
+        PAUSED = 2
     }
 
     private readonly AmqpConnection _connection;
-    private PauseStatus _pauseStatus = PauseStatus.UNPAUSED;
     private readonly string _address;
     private readonly MessageHandler _messageHandler;
     private readonly int _initialCredits;
     private readonly Map _filters;
     private ReceiverLink? _receiverLink;
+    private PauseStatus _pauseStatus = PauseStatus.UNPAUSED;
+    private readonly UnsettledMessageCounter _unsettledMessageCounter = new();
 
     public AmqpConsumer(AmqpConnection connection, string address,
         MessageHandler messageHandler, int initialCredits, Map filters)
@@ -88,7 +93,8 @@ await base.OpenAsync()
 
     private void OnReceiverLinkMessage(IReceiverLink link, Message message)
     {
-        IContext context = new DeliveryContext(link, message);
+        _unsettledMessageCounter.Increment();
+        IContext context = new DeliveryContext(link, message, _unsettledMessageCounter);
         _messageHandler(context, new AmqpMessage(message));
     }
 
@@ -121,9 +127,12 @@ public void Pause()
         }
     }
 
-    public long UnsettledMessageCount()
+    public long UnsettledMessageCount
     {
-        throw new NotImplementedException();
+        get
+        {
+            return _unsettledMessageCounter.Get();
+        }
     }
 
     public void Unpause()

RabbitMQ.AMQP.Client/Impl/AmqpManagement.cs

@@ -389,7 +389,7 @@ internal void CheckResponse(Message sentMessage, int[] expectedResponseCodes, Me
         switch (responseCode)
         {
             case Code409:
-                throw new PreconditionFailedException($"Precondition Fail. Message: {receivedMessage.Body}");
+                throw new PreconditionFailedException($"{receivedMessage.Body}");
         }
 
         // Check if the correlationId is the same as the messageId

RabbitMQ.AMQP.Client/Impl/AmqpPublisher.cs

@@ -1,4 +1,4 @@
-using System.Diagnostics;
+using System.Diagnostics;
 using Amqp;
 using Amqp.Framing;
 using Trace = Amqp.Trace;
@@ -8,14 +8,13 @@ namespace RabbitMQ.AMQP.Client.Impl;
 
 public class AmqpPublisher : AbstractReconnectLifeCycle, IPublisher
 {
-    private SenderLink? _senderLink = null;
-
-    private readonly ManualResetEvent _pausePublishing = new(true);
     private readonly AmqpConnection _connection;
     private readonly TimeSpan _timeout;
     private readonly string _address;
     private readonly int _maxInFlight;
 
+    private SenderLink? _senderLink = null;
+
     public AmqpPublisher(AmqpConnection connection, string address, TimeSpan timeout, int maxInFlight)
     {
         _address = address;
@@ -75,33 +74,6 @@ await base.OpenAsync()
 
     private string Id { get; } = Guid.NewGuid().ToString();
 
-    private void PausePublishing()
-    {
-        _pausePublishing.Reset();
-    }
-
-    private void MaybeResumePublishing()
-    {
-        if (State is State.Open)
-        {
-            // Can be resumed only if the publisher is open and the in-flight messages are less than the max allowed
-            // In case the publisher is closed, closing or reconnecting, the publishing will be paused
-            _pausePublishing.Set();
-        }
-    }
-
-    private void MaybeBackPressure()
-    {
-        if (_currentInFlight >= _maxInFlight)
-        {
-            PausePublishing();
-        }
-        else
-        {
-            MaybeResumePublishing();
-        }
-    }
-
     // TODO: Consider implementing this method with the send method
     // a way to send a batch of messages
 
@@ -114,59 +86,37 @@ private void MaybeBackPressure()
     //     return batch;
     // }
 
-    private int _currentInFlight = 0;
-
-    public Task Publish(IMessage message, OutcomeDescriptorCallback outcomeCallback)
+    public async Task<PublishResult> PublishAsync(IMessage message, CancellationToken cancellationToken = default)
     {
         ThrowIfClosed();
+
+        if (_senderLink is null)
+        {
+            // TODO create "internal bug" exception type?
+            throw new InvalidOperationException("_senderLink is null, report via https://github.com/rabbitmq/rabbitmq-amqp-dotnet-client/issues");
+        }
+
         try
         {
-            _pausePublishing.WaitOne(_timeout);
-            Interlocked.Increment(ref _currentInFlight);
-
-            var nMessage = ((AmqpMessage)message).NativeMessage;
-            _senderLink?.Send(nMessage,
-                (sender, outMessage, outcome, state) =>
-                {
-                    Interlocked.Decrement(ref _currentInFlight);
-                    MaybeBackPressure();
-
-                    if (outMessage == nMessage &&
-                        outMessage.GetEstimatedMessageSize() == nMessage.GetEstimatedMessageSize())
-                    {
-                        if (outcome is Rejected rejected)
-                        {
-                            outcomeCallback(message, new OutcomeDescriptor(rejected.Descriptor.Code,
-                                rejected.Descriptor.ToString(),
-                                OutcomeState.Failed, Utils.ConvertError(rejected?.Error)));
-                        }
-                        else
-                        {
-                            outcomeCallback(message, new OutcomeDescriptor(outcome.Descriptor.Code,
-                                outcome.Descriptor.ToString(),
-                                OutcomeState.Accepted, null));
-                        }
-                    }
-                    else
-                    {
-                        Trace.WriteLine(TraceLevel.Error, $"{ToString()} Message not sent. Killing the process.");
-                        Process.GetCurrentProcess().Kill();
-                    }
-
-                    // is it correct to dispose the message here?
-                    // maybe we should expose a method to dispose the message
-                    nMessage.Dispose();
-                }, this);
+            Debug.Assert(false == _senderLink.IsClosed);
+            Message nativeMessage = ((AmqpMessage)message).NativeMessage;
+            await _senderLink.SendAsync(nativeMessage)
+                .ConfigureAwait(false);
+
+            var publishOutcome = new PublishOutcome(OutcomeState.Accepted, null);
+            return new PublishResult(message, publishOutcome);
+        }
+        catch (AmqpException ex)
+        {
+            var publishOutcome = new PublishOutcome(OutcomeState.Failed, Utils.ConvertError(ex.Error));
+            return new PublishResult(message, publishOutcome);
         }
         catch (Exception e)
         {
             throw new PublisherException($"{ToString()} Failed to publish message, {e}");
         }
-
-        return Task.CompletedTask;
     }
 
-
     public override async Task CloseAsync()
     {
         if (State == State.Closed)
@@ -183,8 +133,6 @@ public override async Task CloseAsync()
                 await _senderLink.CloseAsync()
                     .ConfigureAwait(false);
             }
-
-            _pausePublishing.Dispose();
         }
         catch (Exception e)
         {
@@ -195,9 +143,64 @@ await _senderLink.CloseAsync()
         _connection.Publishers.TryRemove(Id, out _);
     }
 
-
     public override string ToString()
     {
         return $"Publisher{{Address='{_address}', id={Id} ConnectionName='{_connection}', State='{State}'}}";
     }
 }
+
+/*
+Task.Run(() =>
+{
+    // TODO timeout / cancellation
+    // TODO LRB I removed the nativeMessage == outMessage check because outMessage came in as NULL
+    // which I didn't think possible 🤔
+    void OutcomeCallback(ILink sender, Message outMessage, Outcome outcome, object state)
+    {
+        try
+        {
+            Interlocked.Decrement(ref _currentInFlight);
+            MaybeBackPressure();
+
+            PublishOutcome publishOutcome;
+            if (outcome is Rejected rejected)
+            {
+                publishOutcome = new PublishOutcome(rejected.Descriptor.Code, rejected.Descriptor.ToString(),
+                    OutcomeState.Failed, Utils.ConvertError(rejected?.Error));
+
+            }
+            else
+            {
+                publishOutcome = new PublishOutcome(outcome.Descriptor.Code, outcome.Descriptor.ToString(),
+                    OutcomeState.Accepted, null);
+            }
+
+            var result = new PublishResult(message, publishOutcome);
+            publishedTcs.SetResult(result);
+        }
+        catch (Exception ex)
+        {
+            if (false == publishedTcs.TrySetException(ex))
+            {
+                // TODO log it at debug level?
+            }
+        }
+        finally
+        {
+            // is it correct to dispose the message here?
+            // maybe we should expose a method to dispose the message
+            nativeMessage.Dispose();
+        }
+    }
+
+    try
+    {
+        Debug.Assert(false == _senderLink.IsClosed);
+        _senderLink.Send(nativeMessage, OutcomeCallback, this);
+    }
+    catch (Exception ex)
+    {
+        publishedTcs.SetException(ex);
+    }
+}, cancellationToken);
+*/