From 7dea3bb2b3626888e045a1f59a56ad5362ec736c Mon Sep 17 00:00:00 2001
From: Radu Popovici <oncicaradupopovici@gmail.com>
Date: Tue, 26 Jul 2022 11:48:10 +0300
Subject: [PATCH] complete request stream on rusi sub dispose (#234)

Co-authored-by: Radu Popovici <rpopovici@totalsoft.ro>
---
 .../NBB.Messaging.Rusi/RusiMessagingTransport.cs  | 15 ++++++++++++++-
 1 file changed, 14 insertions(+), 1 deletion(-)

diff --git a/src/Messaging/NBB.Messaging.Rusi/RusiMessagingTransport.cs b/src/Messaging/NBB.Messaging.Rusi/RusiMessagingTransport.cs
index 3efb8333..d0a536ba 100644
--- a/src/Messaging/NBB.Messaging.Rusi/RusiMessagingTransport.cs
+++ b/src/Messaging/NBB.Messaging.Rusi/RusiMessagingTransport.cs
@@ -131,7 +131,20 @@ public async Task<IDisposable> SubscribeAsync(string topic, Func<TransportReceiv
                 }
             }, cancellationToken);
 
-            return Task.FromResult<IDisposable>(subscription);
+            return Task.FromResult<IDisposable>(new RusiSubscription(subscription));
+        }
+    }
+
+    internal class RusiSubscription : IDisposable
+    {
+        private readonly AsyncDuplexStreamingCall<SubscribeRequest, ReceivedMessage> _call;
+
+        public RusiSubscription(AsyncDuplexStreamingCall<SubscribeRequest, ReceivedMessage> call) => _call = call;
+
+        public void Dispose()
+        {
+            _call.RequestStream.CompleteAsync().GetAwaiter().GetResult(); //block since the api does not support IAsyncDisposable
+            _call.Dispose();
         }
     }
 }