Eventstream channels (#2101)

* subscribe to eventstream using typed channel

* assert publish

* rename

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

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);
+    }
 }