nats unhandled exception wrapping (#1784)

Author: davte-beijer

src/DotNetCore.CAP.NATS/NATSConsumerClient.cs

@@ -149,8 +149,8 @@ private async void SubscriptionMessageHandler(object? sender, MsgHandlerEventArg
     {
         if (_groupConcurrent > 0)
         {
-            _semaphore.Wait();
-            _ = Task.Run(ConsumeAsync).ConfigureAwait(false);
+            await _semaphore.WaitAsync().ConfigureAwait(false);
+            _ = Task.Run(ConsumeAsync);
         }
         else
         {
@@ -183,21 +183,69 @@ Task ConsumeAsync()
 
     public Task CommitAsync(object? sender)
     {
-        if (sender is Msg msg)
+        try
         {
-            msg.Ack();
+            if (sender is Msg msg)
+            {
+                msg.Ack();
+            }
+        }
+        catch (Exception ex)
+        {
+            OnLogCallback?.Invoke(new LogMessageEventArgs
+            {
+                LogType = MqLogType.AsyncErrorEvent,
+                Reason = $"NATS message ACK failed: {ex}"
+            });
+        }
+        finally
+        {
+            if (_groupConcurrent > 0)
+            {
+                try
+                {
+                    _semaphore.Release();
+                }
+                catch (SemaphoreFullException)
+                {
+                    // ignore (defensive)
+                }
+            }
         }
-        _semaphore.Release();
         return Task.CompletedTask;
     }
 
     public Task RejectAsync(object? sender)
     {
-        if (sender is Msg msg)
+        try
         {
-            msg.Nak();
+            if (sender is Msg msg)
+            {
+                msg.Nak();
+            }
+        }
+        catch (Exception ex)
+        {
+            OnLogCallback?.Invoke(new LogMessageEventArgs
+            {
+                LogType = MqLogType.AsyncErrorEvent,
+                Reason = $"NATS message NAK failed: {ex}"
+            });
+        }
+        finally
+        {
+            if (_groupConcurrent > 0)
+            {
+                try
+                {
+                    _semaphore.Release();
+                }
+                catch (SemaphoreFullException)
+                {
+                    // ignore (defensive)
+                }
+            }
         }
-        _semaphore.Release();
         return Task.CompletedTask;
     }