handler messaging processing error

.gitignore

@@ -1,19 +1,19 @@
-################################################################################
-# This .gitignore file was automatically created by Microsoft(R) Visual Studio.
-################################################################################
-
-/.vs
-/.idea
-/artifacts
-*/**/bin
-*/**/obj
-**/packages
-*/**/*.csproj.user
+################################################################################
+# This .gitignore file was automatically created by Microsoft(R) Visual Studio.
+################################################################################
+
+/.vs
+/.idea
+/artifacts
+*/**/bin
+*/**/obj
+**/packages
+*/**/*.csproj.user
 TestResults
 */**/BenchmarkDotNet.Artifacts
 **/*.DotSettings.user
 */**/appsettings.Development.json
 *.user
-*.suo
-.fake
-.ionide
+*.suo
+.fake
+.ionide

NBB.Messaging.slnf

@@ -6,6 +6,7 @@
  "src\\Messaging\\NBB.Messaging.Effects\\NBB.Messaging.Effects.csproj",
  "src\\Messaging\\NBB.Messaging.Host\\NBB.Messaging.Host.csproj",
  "src\\Messaging\\NBB.Messaging.InProcessMessaging\\NBB.Messaging.InProcessMessaging.csproj",
+ "src\\Messaging\\NBB.Messaging.JetStream\\NBB.Messaging.JetStream.csproj",
  "src\\Messaging\\NBB.Messaging.MultiTenancy\\NBB.Messaging.MultiTenancy.csproj",
  "src\\Messaging\\NBB.Messaging.Nats\\NBB.Messaging.Nats.csproj",
  "src\\Messaging\\NBB.Messaging.Noop\\NBB.Messaging.Noop.csproj",

src/Messaging/NBB.Messaging.Abstractions/IMessageBusSubscriber.cs

@@ -1,20 +1,20 @@
 // Copyright (c) TotalSoft.
 // This source code is licensed under the MIT license.
 
-using System;
-using System.Threading;
-using System.Threading.Tasks;
-
-namespace NBB.Messaging.Abstractions
-{
- public interface IMessageBusSubscriber
- {
- Task<IDisposable> SubscribeAsync<TMessage>(Func<MessagingEnvelope<TMessage>, Task> handler,
- MessagingSubscriberOptions options = null, CancellationToken cancellationToken = default);
-
-
- Task<IDisposable> SubscribeAsync(Func<MessagingEnvelope, Task> handler,
- MessagingSubscriberOptions options = null, CancellationToken cancellationToken = default)
- => SubscribeAsync<object>(handler, options, cancellationToken);
- }
-}
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace NBB.Messaging.Abstractions
+{
+ public interface IMessageBusSubscriber
+ {
+ Task<IDisposable> SubscribeAsync<TMessage>(Func<MessagingEnvelope<TMessage>, Task> handler,
+ MessagingSubscriberOptions options = null, CancellationToken cancellationToken = default);
+
+
+ Task<IDisposable> SubscribeAsync(Func<MessagingEnvelope, Task> handler,
+ MessagingSubscriberOptions options = null, CancellationToken cancellationToken = default)
+ => SubscribeAsync<object>(handler, options, cancellationToken);
+ }
+}

src/Messaging/NBB.Messaging.Abstractions/IMessagingTransport.cs

@@ -18,7 +18,7 @@ public interface IMessagingTransport
  /// <param name="options">Subscription options</param>
  /// <param name="cancellationToken"></param>
  /// <returns>An object that when disposed unsubscribes the handler from the topic</returns>
- Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task> handler,
+ Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task<PipelineResult>> handler,
  SubscriptionTransportOptions options = null, CancellationToken cancellationToken = default);
 
  /// <summary>
@@ -45,6 +45,11 @@ public record TransportSendContext(
 
  public record TransportReceiveContext(TransportReceivedData ReceivedData);
 
+ public record PipelineResult(bool Success, string Error)
+ {
+ public static PipelineResult SuccessResult = new PipelineResult(true, null);
+ }
+
 
  public abstract record TransportReceivedData
  {

src/Messaging/NBB.Messaging.Abstractions/MessageBusSubscriber.cs

@@ -36,11 +36,12 @@ public class MessageBusSubscriber : IMessageBusSubscriber
 
  var topicName = _topicRegistry.GetTopicForName(topicNameWithoutPrefix);
 
- async Task MsgHandler(TransportReceiveContext receiveContext)
+ async Task<PipelineResult> MsgHandler(TransportReceiveContext receiveContext)
  {
  _logger.LogDebug("Messaging subscriber received message from subject {Subject}", topicName);
 
  MessagingEnvelope<TMessage> messageEnvelope = null;
+
  try
  {
  messageEnvelope = receiveContext.ReceivedData switch
@@ -51,19 +52,30 @@ async Task MsgHandler(TransportReceiveContext receiveContext)
  => new MessagingEnvelope<TMessage>(headers, _messageSerDes.DeserializePayload<TMessage>(payloadBytes, headers, options?.SerDes)),
  _ => throw new Exception("Invalid received message data")
  };
+ }
+ catch (Exception ex)
+ {
+ _logger.LogError(ex, "Deserialization error on topic {Subject}.", topicName);
+
+ if (messageEnvelope != null)
+ _deadLetterQueue.Push(messageEnvelope, topicName, ex);
+ else
+ _deadLetterQueue.Push(receiveContext.ReceivedData, topicName, ex);
 
+ return new PipelineResult(false, ex.Message);
+ }
+
+ try
+ {
  await handler(messageEnvelope);
+ return PipelineResult.SuccessResult;
  }
  catch (Exception ex)
  {
  _logger.LogError(ex, "Messaging subscriber encountered an error when handling a message from subject {Subject}.",
  topicName);
 
- if (messageEnvelope != null)
- _deadLetterQueue.Push(messageEnvelope, topicNameWithoutPrefix, ex);
- else
- _deadLetterQueue.Push(receiveContext.ReceivedData, topicNameWithoutPrefix, ex);
-
+ return new PipelineResult(false, ex.Message);
  }
  }
 

src/Messaging/NBB.Messaging.Host/MessagingPipeline/ExceptionHandlingMiddleware.cs

@@ -48,10 +48,12 @@ public async Task Invoke(MessagingContext context, CancellationToken cancellatio
  "An unhandled exception has occurred while processing message of type {MessageType}.",
  context.MessagingEnvelope.Payload.GetType().GetPrettyName());
 
- Activity.Current?.SetException(ex);
- Activity.Current?.SetStatus(Status.Error);
+ //Activity.Current?.SetException(ex);
+ //Activity.Current?.SetStatus(Status.Error);
 
  _deadLetterQueue.Push(context.MessagingEnvelope, context.TopicName, ex);
+
+ throw;
  }
  finally
  {

src/Messaging/NBB.Messaging.InProcessMessaging/Internal/InProcessMessagingTransport.cs

@@ -23,7 +23,7 @@ public InProcessMessagingTransport(IStorage storage, ILogger<InProcessMessagingT
 
  public event TransportErrorHandler OnError;
 
- public async Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task> handler,
+ public async Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task<PipelineResult>> handler,
  SubscriptionTransportOptions options = null,
  CancellationToken cancellationToken = default)
  {

src/Messaging/NBB.Messaging.JetStream/JetStreamMessagingTransport.cs

@@ -34,10 +34,8 @@ public Task PublishAsync(string topic, TransportSendContext sendContext, Cancell
  });
  }
 
- public Task<IDisposable> SubscribeAsync(string topic,
- Func<TransportReceiveContext, Task> handler,
- SubscriptionTransportOptions options = null,
- CancellationToken cancellationToken = default)
+ public Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task<PipelineResult>> handler,
+ SubscriptionTransportOptions options = null, CancellationToken cancellationToken = default)
  {
 
  IDisposable consumer = null;

src/Messaging/NBB.Messaging.Nats/StanMessagingTransport.cs

@@ -22,7 +22,7 @@ public StanMessagingTransport(StanConnectionProvider stanConnectionManager, IOpt
  _natsOptions = natsOptions;
  }
 
- public Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task> handler,
+ public Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task<PipelineResult>> handler,
  SubscriptionTransportOptions options = null,
  CancellationToken cancellationToken = default)
  {

src/Messaging/NBB.Messaging.Noop/NoopMessagingTransport.cs

@@ -10,7 +10,7 @@ namespace NBB.Messaging.Noop
 {
  public class NoopMessagingTransport : IMessagingTransport
  {
- public Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task> handler,
+ public Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task<PipelineResult>> handler,
  SubscriptionTransportOptions options = null,
  CancellationToken cancellationToken = default)
  {

src/Messaging/NBB.Messaging.OpenTelemetry/Publisher/OpenTelemetryPublisherDecorator.cs

@@ -47,7 +47,12 @@ void NewCustomizer(MessagingEnvelope outgoingEnvelope)
 
  var formattedTopicName = _topicRegistry.GetTopicForName(options.TopicName) ??
  _topicRegistry.GetTopicForMessageType(message.GetType());
- var operationName = $"{message.GetType().GetPrettyName()} send";
+
+ var prettyName = message.GetType().GetPrettyName();
+ if (prettyName.Contains("AnonymousType"))
+ prettyName = formattedTopicName;
+
+ var operationName = $"{prettyName} send";
 
  using var activity = activitySource.StartActivity(operationName, ActivityKind.Producer);
 

src/Messaging/NBB.Messaging.OpenTelemetry/Subscriber/OpenTelemetrySubscriberDecorator.cs

@@ -1,21 +1,16 @@
 // Copyright (c) TotalSoft.
 // This source code is licensed under the MIT license.
 
+using NBB.Core.Abstractions;
 using NBB.Messaging.Abstractions;
-using OpenTelemetry.Trace;
 using OpenTelemetry;
+using OpenTelemetry.Context.Propagation;
+using OpenTelemetry.Trace;
 using System;
-using System.Collections.Generic;
 using System.Diagnostics;
 using System.Linq;
-using System.Reflection;
-using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using OpenTelemetry.Context.Propagation;
-using NBB.Core.Abstractions;
-using NBB.Messaging.OpenTelemetry.Publisher;
-using System.Reflection.PortableExecutable;
 
 namespace NBB.Messaging.OpenTelemetry.Subscriber
 {

src/Messaging/NBB.Messaging.Rusi/RusiMessagingTransport.cs

@@ -59,7 +59,7 @@ internal class RusiMessagingTransport : IMessagingTransport, ITransportMonitor
  await _client.PublishAsync(request);
  }
 
- public async Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task> handler,
+ public async Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task<PipelineResult>> handler,
  SubscriptionTransportOptions options = null, CancellationToken cancellationToken = default)
  {
  var transport = options ?? SubscriptionTransportOptions.Default;
@@ -108,8 +108,8 @@ await foreach (var msg in subscription.ResponseStream.ReadAllAsync(cancellationT
  {
  try
  {
- await handler(receiveContext);
- await ackChannel.Writer.WriteAsync(new() { MessageId = msg.Id });
+ var result = await handler(receiveContext);
+ await ackChannel.Writer.WriteAsync(new() { MessageId = msg.Id, Error = result.Error });
  }
  catch (Exception ex)
  {