Merge pull request #73 from ChrisPulman/FixForMultiTopics

Fix inability to Subscribe to multiple topics from the same client
ChrisPulman · Sep 5, 2024 · 511efb1 · 511efb1
2 parents d92dd25 + c6374f6
commit 511efb1
Show file tree

Hide file tree

Showing 7 changed files with 37 additions and 27 deletions.
diff --git a/Version.json b/Version.json
@@ -1,6 +1,6 @@
 {
     "$schema": "https://raw.githubusercontent.com/dotnet/Nerdbank.GitVersioning/master/src/NerdBank.GitVersioning/version.schema.json",
-    "version": "2.1.1",
+    "version": "2.2.0",
     "publicReleaseRefSpec": [
         "^refs/heads/master$",
         "^refs/heads/main$"

diff --git a/global.json b/global.json
diff --git a/src/MQTTnet.Rx.Client.TestApp/Program.cs b/src/MQTTnet.Rx.Client.TestApp/Program.cs
@@ -12,7 +12,8 @@ namespace MQTTnet.Rx.Client.TestApp
     internal static class Program
     {
         private static readonly Subject<(string topic, string payload)> _message = new();
-        private static CompositeDisposable _disposables = new();
+        private static readonly JsonSerializerOptions _jsonOptions = new() { WriteIndented = true };
+        private static CompositeDisposable _disposables = [];
 
         /// <summary>
         /// Defines the entry point of the application.
@@ -128,9 +129,9 @@ private static void DiscoverTopicsManagedClient()
                 .Subscribe(r =>
                 {
                     Console.Clear();
-                    foreach (var topic in r)
+                    foreach (var (topic, lastSeen) in r)
                     {
-                        Console.WriteLine($"{topic.Topic} Last Seen: {topic.LastSeen}");
+                        Console.WriteLine($"{topic} Last Seen: {lastSeen}");
                     }
                 }));
             WaitForExit();
@@ -161,18 +162,15 @@ private static void WaitForExit(string? message = null, bool clear = true)
             }
 
             _disposables.Dispose();
-            _disposables = new();
+            _disposables = [];
         }
 
         private static TObject DumpToConsole<TObject>(this TObject @object)
         {
             var output = "NULL";
             if (@object != null)
             {
-                output = JsonSerializer.Serialize(@object, new JsonSerializerOptions
-                {
-                    WriteIndented = true
-                });
+                output = JsonSerializer.Serialize(@object, _jsonOptions);
             }
 
             Console.WriteLine($"[{@object?.GetType().Name}]:\r\n{output}");

diff --git a/src/MQTTnet.Rx.Client/CreateObservable.cs b/src/MQTTnet.Rx.Client/CreateObservable.cs
@@ -9,7 +9,7 @@ namespace MQTTnet.Rx.Client;
 /// <summary>
 /// Mqtt Extensions.
 /// </summary>
-public static class CreateObservable
+internal static class CreateObservable
 {
     internal static IObservable<T> FromAsyncEvent<T>(Action<Func<T, Task>> addHandler, Action<Func<T, Task>> removeHandler) =>
         Observable.Create<T>(observer =>

diff --git a/src/MQTTnet.Rx.Client/MqttdSubscribeExtensions.cs b/src/MQTTnet.Rx.Client/MqttdSubscribeExtensions.cs
@@ -15,9 +15,9 @@ namespace MQTTnet.Rx.Client;
 /// </summary>
 public static class MqttdSubscribeExtensions
 {
-    private static readonly Dictionary<string, IObservable<object?>> _dictJsonValues = new();
-    private static readonly Dictionary<IManagedMqttClient, List<(string topic, int count)>> _managedSubscribeToTopicClients = new();
-    private static readonly Dictionary<IMqttClient, List<(string topic, int count)>> _unmanagedSubscribeToTopicClients = new();
+    private static readonly Dictionary<string, IObservable<object?>> _dictJsonValues = [];
+    private static readonly Dictionary<IManagedMqttClient, List<(string topic, int count)>> _managedSubscribeToTopicClients = [];
+    private static readonly Dictionary<IMqttClient, List<(string topic, int count)>> _unmanagedSubscribeToTopicClients = [];
 
     /// <summary>
     /// Converts to dictionary.
@@ -126,16 +126,20 @@ public static IObservable<MqttApplicationMessageReceivedEventArgs> SubscribeToTo
             disposable.Add(client.Subscribe(async c =>
             {
                 mqttClient = c;
-                disposable.Add(mqttClient.ApplicationMessageReceived().Subscribe(observer));
                 if (!_unmanagedSubscribeToTopicClients.TryGetValue(mqttClient, out var value))
                 {
-                    value = new List<(string topic, int count)>(new[] { (topic, 0) });
+                    value = new([(topic, 0)]);
                     _unmanagedSubscribeToTopicClients.Add(mqttClient, value);
                 }
+                else if (!value.Any(x => x.topic == topic))
+                {
+                    value.Add((topic, 0));
+                }
 
                 var check = value.Find(x => x.topic == topic);
                 if (!EqualityComparer<(string topic, int count)>.Default.Equals(check, default))
                 {
+                    disposable.Add(mqttClient.ApplicationMessageReceived().Where(x => x.ApplicationMessage.Topic == topic).Subscribe(observer));
                     check.count++;
                     if (check.count == 1)
                     {
@@ -309,24 +313,28 @@ public static IObservable<MqttApplicationMessageReceivedEventArgs> SubscribeToTo
             disposable.Add(client.Subscribe(async c =>
             {
                 mqttClient = c;
-                disposable.Add(mqttClient.ApplicationMessageReceived().Subscribe(observer));
                 if (!_managedSubscribeToTopicClients.TryGetValue(mqttClient, out var value))
                 {
-                    value = new List<(string topic, int count)>(new[] { (topic, 0) });
+                    value = new([(topic, 0)]);
                     _managedSubscribeToTopicClients.Add(mqttClient, value);
                 }
+                else if (!value.Any(x => x.topic == topic))
+                {
+                    value.Add((topic, 0));
+                }
 
                 var check = value.Find(x => x.topic == topic);
                 if (!EqualityComparer<(string topic, int count)>.Default.Equals(check, default))
                 {
+                    disposable.Add(mqttClient.ApplicationMessageReceived().Where(x => x.ApplicationMessage.Topic == topic).Subscribe(observer));
                     check.count++;
                     if (check.count == 1)
                     {
                         var mqttSubscribeOptions = Create.MqttFactory.CreateTopicFilterBuilder()
                                             .WithTopic(topic)
                                             .Build();
 
-                        await mqttClient.SubscribeAsync(new[] { mqttSubscribeOptions });
+                        await mqttClient.SubscribeAsync([mqttSubscribeOptions]);
                     }
                 }
             }));
@@ -342,7 +350,7 @@ public static IObservable<MqttApplicationMessageReceivedEventArgs> SubscribeToTo
                                 check.count--;
                                 if (check.count == 0)
                                 {
-                                    await mqttClient.UnsubscribeAsync(new[] { topic }).ConfigureAwait(false);
+                                    await mqttClient.UnsubscribeAsync([topic]).ConfigureAwait(false);
                                 }
                             }
                         }

diff --git a/src/MQTTnet.Rx.sln b/src/MQTTnet.Rx.sln
@@ -1,5 +1,5 @@
 Microsoft Visual Studio Solution File, Format Version 12.00
-# Visual Studio Version 17
+# 17
 VisualStudioVersion = 17.7.34003.232
 MinimumVisualStudioVersion = 10.0.40219.1
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "MQTTnet.Rx.ABPlc", "MQTTnet.Rx.ABPlc\MQTTnet.Rx.ABPlc.csproj", "{5DB1DAEB-5A4E-436C-8D23-AC185DD67276}"
@@ -9,7 +9,6 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "SolutionConfig", "SolutionC
 		..\.editorconfig = ..\.editorconfig
 		..\.gitignore = ..\.gitignore
 		..\Directory.Build.props = ..\Directory.Build.props
-		..\global.json = ..\global.json
 		..\LICENSE = ..\LICENSE
 		..\README.md = ..\README.md
 		..\stylecop.json = ..\stylecop.json

diff --git a/src/TestSingleClient/Program.cs b/src/TestSingleClient/Program.cs
@@ -50,11 +50,19 @@
           var s4 = obsClient2.SubscribeToTopic("FromMilliseconds")
                    .Subscribe(r => Console.WriteLine($"\tCLIENT S4: {r.ReasonCode} [{r.ApplicationMessage.Topic}] value : {r.ApplicationMessage.ConvertPayloadToString()}"));
 
+          var s5 = obsClient1.SubscribeToTopic("FromMilliseconds1")
+                   .Subscribe(r => Console.WriteLine($"\tCLIENT S5: {r.ReasonCode} [{r.ApplicationMessage.Topic}] value : {r.ApplicationMessage.ConvertPayloadToString()}"));
+
           Subject<(string topic, string payload)> message = new();
 
           sub.Disposable.Add(Observable.Interval(TimeSpan.FromMilliseconds(1000)).Subscribe(i => message.OnNext(("FromMilliseconds", "{" + $"payload: {i}" + "}"))));
           sub.Disposable.Add(obsClient1.PublishMessage(message).Subscribe());
 
+          Subject<(string topic, string payload)> message1 = new();
+
+          sub.Disposable.Add(Observable.Interval(TimeSpan.FromMilliseconds(1000)).Subscribe(i => message.OnNext(("FromMilliseconds1", "{" + $"payload: {i}" + "}"))));
+          sub.Disposable.Add(obsClient1.PublishMessage(message1).Subscribe());
+
           await Task.Delay(3000);
           s2.Dispose();
           Console.WriteLine("Dispose S2 ---------------");
@@ -68,6 +76,9 @@
           Console.Read();
           s3.Dispose();
           Console.WriteLine("Dispose S3 ---------------");
+          Console.Read();
+          s5.Dispose();
+          Console.WriteLine("Dispose S5 ---------------");
       });
 Console.WriteLine("Press 'Escape' or 'Q' to exit.");