From 1ea5d4f9f63bb35688acc0f2c6a8ffb98735d559 Mon Sep 17 00:00:00 2001
From: Roger Johansson <rogeralsing@gmail.com>
Date: Wed, 28 Feb 2024 13:32:52 +0100
Subject: [PATCH] Eventstream channels (#2101)

* subscribe to eventstream using typed channel

* assert publish

* rename
---
 src/Proto.Actor/EventStream/EventStream.cs  | 25 +++++++++++++++++++++
 tests/Proto.Actor.Tests/EventStreamTests.cs | 18 +++++++++++++++
 2 files changed, 43 insertions(+)

diff --git a/src/Proto.Actor/EventStream/EventStream.cs b/src/Proto.Actor/EventStream/EventStream.cs
index d4b8553445..d8158c773e 100644
--- a/src/Proto.Actor/EventStream/EventStream.cs
+++ b/src/Proto.Actor/EventStream/EventStream.cs
@@ -119,6 +119,31 @@ public EventStreamSubscription<T> Subscribe(Channel<T> channel, IDispatcher? dis
 
         return sub;
     }
+    
+    /// <summary>
+    ///     Subscribe to messages and yields the result onto a Channel
+    /// </summary>
+    /// <param name="channel">a Channel which receives the event</param>
+    /// <param name="dispatcher">Optional: the dispatcher, will use <see cref="Dispatchers.SynchronousDispatcher" /> by default</param>
+    /// <returns>A new subscription that can be used to unsubscribe</returns>
+    public EventStreamSubscription<T> Subscribe<TMsg>(Channel<TMsg> channel, IDispatcher? dispatcher = null) where TMsg:T
+    {
+        var sub = new EventStreamSubscription<T>(
+            this,
+            dispatcher ?? Dispatchers.SynchronousDispatcher,
+            async x =>
+            {
+                if (x is TMsg tc)
+                {
+                    await channel.Writer.WriteAsync(tc).ConfigureAwait(false);
+                }
+            }
+        );
+
+        _subscriptions.TryAdd(sub.Id, sub);
+
+        return sub;
+    }
 
     /// <summary>
     ///     Subscribe to messages with an asynchronous handler
diff --git a/tests/Proto.Actor.Tests/EventStreamTests.cs b/tests/Proto.Actor.Tests/EventStreamTests.cs
index 6daa2913fe..adcd1622a1 100644
--- a/tests/Proto.Actor.Tests/EventStreamTests.cs
+++ b/tests/Proto.Actor.Tests/EventStreamTests.cs
@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 using Proto.Mailbox;
 using Xunit;
@@ -80,4 +81,21 @@ public async Task EventStream_CanSubscribeToSpecificEventTypes_Async()
 
         eventStream.Publish("hello");
     }
+
+    [Fact]
+    public async Task EventStream_CanSubscribeUsingChannel()
+    {
+        var system = new ActorSystem();
+        await using var _ = system;
+        var eventStream = system.EventStream;
+        
+        var channel = Channel.CreateUnbounded<string>();
+        eventStream.Subscribe(channel);
+        eventStream.Publish(123);
+        eventStream.Publish(false);
+        eventStream.Publish("hello");
+
+        var res = await channel.Reader.ReadAsync();
+        Assert.Equal("hello",res);
+    }
 }
\ No newline at end of file