response streaming working

coronabytes · Sep 14, 2024 · 7ed987e · 7ed987e
1 parent 12be938
commit 7ed987e
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/Core.ServiceMesh/Internal/ServiceMeshWorker.cs b/Core.ServiceMesh/Internal/ServiceMeshWorker.cs
@@ -12,6 +12,7 @@
 using OpenTelemetry;
 using OpenTelemetry.Context.Propagation;
 using OpenTelemetry.Trace;
+using static System.Runtime.InteropServices.JavaScript.JSType;
 
 namespace Core.ServiceMesh.Internal;
 
@@ -443,8 +444,11 @@ private async ValueTask AsyncEnumerableWrapper<T>(string sub, IAsyncEnumerable<T
     {
         await foreach (var val in stream)
         {
-            await nats.PublishAsync(sub, options.Serialize(val!, true));
+            var data = options.Serialize(val!, true);
+            await nats.PublishAsync(sub, data);
         }
+
+        await nats.PublishAsync(sub, Array.Empty<byte>());
     }
 
     private async Task DurableWorker(CancellationToken stoppingToken)
@@ -609,11 +613,13 @@ public async IAsyncEnumerable<T> StreamAsync<T>(string subject, object[] args, T
 
         var subId = Guid.NewGuid().ToString("N");
 
-        headers["reply-sub-id"] = subId;
+        headers["return-sub-id"] = subId;
         await nats.PublishAsync(subject, body, headers: headers);
 
         await foreach (var msg in nats.SubscribeAsync<byte[]>(subId))
         {
+            if (msg.Data == null)
+                yield break;
             yield return (T)options.Deserialize(msg.Data!, typeof(T), true)!;
         }
     }