Add ITransportMonitor implementation for InProcess messaging (#201)

osstotalsoft · Jan 12, 2022 · 0f909cc · 0f909cc
1 parent 433a51b
commit 0f909cc
Show file tree

Hide file tree

Showing 2 changed files with 13 additions and 4 deletions.
diff --git a/src/Messaging/NBB.Messaging.InProcessMessaging/Extensions/DependencyInjectionExtensions.cs b/src/Messaging/NBB.Messaging.InProcessMessaging/Extensions/DependencyInjectionExtensions.cs
@@ -11,8 +11,11 @@ public static class DependencyInjectionExtensions
     {
         public static void AddInProcessTransport(this IServiceCollection services)
         {
-            services.AddSingleton<IStorage, Storage>();
-            services.AddSingleton<IMessagingTransport, InProcessMessagingTransport>();
+            services.AddSingleton<IStorage, Storage>();
+
+            services.AddSingleton<InProcessMessagingTransport>();
+            services.AddSingleton<ITransportMonitor>(sp => sp.GetRequiredService<InProcessMessagingTransport>());
+            services.AddSingleton<IMessagingTransport>(sp => sp.GetRequiredService<InProcessMessagingTransport>());
         }
     }
 }
diff --git a/src/Messaging/NBB.Messaging.InProcessMessaging/Internal/InProcessMessagingTransport.cs b/src/Messaging/NBB.Messaging.InProcessMessaging/Internal/InProcessMessagingTransport.cs
@@ -10,7 +10,7 @@
 
 namespace NBB.Messaging.InProcessMessaging.Internal
 {
-    public class InProcessMessagingTransport : IMessagingTransport
+    public class InProcessMessagingTransport : IMessagingTransport, ITransportMonitor
     {
         private readonly IStorage _storage;
         private readonly ILogger<InProcessMessagingTransport> _logger;
@@ -21,6 +21,8 @@ public InProcessMessagingTransport(IStorage storage, ILogger<InProcessMessagingT
             _logger = logger;
         }
 
+        public event TransportErrorHandler OnError;
+
         public async Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiveContext, Task> handler,
             SubscriptionTransportOptions options = null,
             CancellationToken cancellationToken = default)
@@ -38,6 +40,9 @@ await _storage.AddSubscription(topic, async msg =>
                     _logger.LogError(
                         "InProcessMessagingTopicSubscriber encountered an error when handling a message from topic {TopicName}.\n {Error}",
                         topic, ex);
+
+                    OnError?.Invoke(ex);
+
                     //TODO: push to DLQ
                 }
             }, cancellationToken);
@@ -61,10 +66,11 @@ public Task PublishAsync(string topic, TransportSendContext sendContext, Cancell
             return Task.CompletedTask;
         }
 
-        private class Subscription : IDisposable
+        private sealed class Subscription : IDisposable
         {
             public void Dispose()
             {
+                // Nothing to dispose
             }
         }
     }