From 80764f2b1605c834a643dd1565044aba28b378c9 Mon Sep 17 00:00:00 2001
From: Dero <devel@deroin.de>
Date: Wed, 1 Apr 2026 12:47:44 +0200
Subject: [PATCH] Added support for unsubscribing previously subscribed from
 topics

---
 source/Interfaces/IStompClient.cs |  4 ++--
 source/StompClient.cs             | 35 +++++++++++++++++++++++++------
 2 files changed, 31 insertions(+), 8 deletions(-)

diff --git a/source/Interfaces/IStompClient.cs b/source/Interfaces/IStompClient.cs
index 7103110..799fff0 100644
--- a/source/Interfaces/IStompClient.cs
+++ b/source/Interfaces/IStompClient.cs
@@ -21,8 +21,8 @@ public interface IStompClient : IDisposable
         Task ConnectAsync(IDictionary<string, string> headers);
         Task SendAsync(object body, string destination, IDictionary<string, string> headers);
         Task SendAsync(string body, string destination, IDictionary<string, string> headers);
-        Task SubscribeAsync<T>(string topic, IDictionary<string, string> headers, EventHandler<T> handler);
-        Task SubscribeAsync(string topic, IDictionary<string, string> headers, EventHandler<StompMessage> handler);
+        Task<StompUnsubscribeCallback> SubscribeAsync<T>(string topic, IDictionary<string, string> headers, EventHandler<T> handler);
+        Task<StompUnsubscribeCallback> SubscribeAsync(string topic, IDictionary<string, string> headers, EventHandler<StompMessage> handler);
         Task AckAsync(string id, string transaction = null);
         Task NackAsync(string id, string transaction = null);
         Task DisconnectAsync();
diff --git a/source/StompClient.cs b/source/StompClient.cs
index c83e304..e7259bc 100644
--- a/source/StompClient.cs
+++ b/source/StompClient.cs
@@ -3,6 +3,7 @@
 using System.Text;
 using System.Threading.Tasks;
 using System.Net.WebSockets;
+using System.Threading;
 using Netina.Stomp.Client.Interfaces;
 using Netina.Stomp.Client.Messages;
 using Netina.Stomp.Client.Utils;
@@ -12,6 +13,8 @@
 
 namespace Netina.Stomp.Client
 {
+    public delegate Task StompUnsubscribeCallback();
+
     public class StompClient : IStompClient
     {
         public event EventHandler<StompMessage> OnConnect;
@@ -27,6 +30,7 @@ public class StompClient : IStompClient
         private readonly StompMessageSerializer _stompSerializer = new StompMessageSerializer();
         private readonly IDictionary<string, EventHandler<StompMessage>> _subscribers = new Dictionary<string, EventHandler<StompMessage>>();
         private readonly IDictionary<string, string> _connectingHeaders = new Dictionary<string, string>();
+        private int _subscriptionIdSequence;
 
         /// <summary>
         /// StompClient Ctor
@@ -120,21 +124,40 @@ public async Task SendAsync(string body, string destination, IDictionary<string,
             await _socket.SendInstant(_stompSerializer.Serialize(connectMessage));
         }
 
-        public async Task SubscribeAsync<T>(string topic, IDictionary<string, string> headers, EventHandler<T> handler)
+        public async Task<StompUnsubscribeCallback> SubscribeAsync<T>(string topic, IDictionary<string, string> headers, EventHandler<T> handler)
         {
-            await SubscribeAsync(topic, headers, (sender, message) => handler(this, JsonConvert.DeserializeObject<T>(message.Body)));
+            return await SubscribeAsync(topic, headers, (sender, message) => handler(this, JsonConvert.DeserializeObject<T>(message.Body)));
         }
 
-        public async Task SubscribeAsync(string topic, IDictionary<string, string> headers, EventHandler<StompMessage> handler)
+        public async Task<StompUnsubscribeCallback> SubscribeAsync(string topic, IDictionary<string, string> headers, EventHandler<StompMessage> handler)
         {
             if (StompState != StompConnectionState.Open)
                 await Reconnect();
 
-            headers.Add("destination", topic);
-            headers.Add("id", $"sub-{_subscribers.Count}");
-            var subscribeMessage = new StompMessage(StompCommand.Subscribe, headers);
+            Dictionary<string, string> headersCopy = new Dictionary<string, string>(headers);
+
+            string subscriptionId = $"sub-{Interlocked.Increment(ref _subscriptionIdSequence)}";
+            headersCopy.Add("destination", topic);
+            headersCopy.Add("id", subscriptionId);
+            var subscribeMessage = new StompMessage(StompCommand.Subscribe, headersCopy);
             await _socket.SendInstant(_stompSerializer.Serialize(subscribeMessage));
             _subscribers.Add(topic, handler);
+
+            return () => UnsubscribeAsync(subscriptionId, topic, headers);
+        }
+
+        private async Task UnsubscribeAsync(string subscriptionId, string topic, IDictionary<string, string> headers) {
+            if (StompState != StompConnectionState.Open) {
+                await Reconnect();
+            }
+
+            Dictionary<string, string> headersCopy = new Dictionary<string, string>(headers);
+
+            headersCopy.Add("destination", topic);
+            headersCopy.Add("id", subscriptionId);
+
+            await _socket.SendInstant(_stompSerializer.Serialize(new StompMessage(StompCommand.Unsubscribe, headersCopy)));
+            _subscribers.Remove(topic);
         }
 
         public async Task AckAsync(string id, string transaction = null)