Eventstream channels

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