Actor test

src/Squidex.Infrastructure.GetEventStore/CQRS/Events/GetEventStoreSubscription.cs

@@ -12,15 +12,19 @@ using System.Collections.Generic;
 using System.Linq;
 using System.Net;
 using System.Net.Sockets;
-using System.Threading;
 using System.Threading.Tasks;
 using EventStore.ClientAPI;
 using EventStore.ClientAPI.Exceptions;
 using EventStore.ClientAPI.Projections;
+using Squidex.Infrastructure.Actors;
+using Squidex.Infrastructure.CQRS.Events.Actors.Messages;
+using Squidex.Infrastructure.Tasks;
+
+#pragma warning disable SA1401 // Fields must be private
 
 namespace Squidex.Infrastructure.CQRS.Events
 {
-    internal sealed class GetEventStoreSubscription : DisposableObjectBase, IEventSubscription
+    internal sealed class GetEventStoreSubscription : Actor, IEventSubscription
     {
         private const int ReconnectWindowMax = 5;
         private const int ReconnectWaitMs = 1000;
@@ -31,13 +35,26 @@ namespace Squidex.Infrastructure.CQRS.Events
         private readonly string streamName;
         private readonly string prefix;
         private readonly string projectionHost;
-        private readonly ReaderWriterLockSlim connectionLock = new ReaderWriterLockSlim();
         private readonly Queue<DateTime> reconnectTimes = new Queue<DateTime>();
-        private readonly CancellationTokenSource disposeToken = new CancellationTokenSource();
-        private Func<StoredEvent, Task> publishNext;
-        private Func<Exception, Task> publishError;
-        private EventStoreCatchUpSubscription internalSubscription;
+        private EventStoreCatchUpSubscription subscription;
         private long? position;
+        private IActor parent;
+
+        private sealed class ConnectMessage : IMessage
+        {
+        }
+
+        private sealed class ConnectionFailedMessage : IMessage
+        {
+            public Exception Exception;
+        }
+
+        private sealed class ReceiveESEventMessage : IMessage
+        {
+            public ResolvedEvent Event;
+
+            public EventStoreCatchUpSubscription Subscription;
+        }
 
         public GetEventStoreSubscription(IEventStoreConnection connection, string streamFilter, string position, string prefix, string projectionHost)
         {
@@ -50,170 +67,93 @@ namespace Squidex.Infrastructure.CQRS.Events
             streamName = $"by-{prefix.Simplify()}-{streamFilter.Simplify()}";
         }
 
-        protected override void DisposeObject(bool disposing)
+        protected override Task OnStop()
         {
-            if (disposing)
-            {
-                disposeToken.Cancel();
+            subscription?.Stop();
 
-                try
-                {
-                    connectionLock.EnterWriteLock();
-
-                    internalSubscription?.Stop();
-                    internalSubscription = null;
-                }
-                finally
-                {
-                    connectionLock.ExitWriteLock();
-                }
-            }
+            return TaskHelper.Done;
         }
 
-        public async Task SubscribeAsync(Func<StoredEvent, Task> onNext, Func<Exception, Task> onError = null)
+        protected override async Task OnError(Exception exception)
         {
-            Guard.NotNull(onNext, nameof(onNext));
-
-            if (publishNext != null)
-            {
-                throw new InvalidOperationException("An handler has already been registered.");
-            }
-
-            publishNext = onNext;
-            publishError = onError;
-
-            await CreateProjectionAsync();
-
-            try
+            if (parent != null)
             {
-                connectionLock.EnterWriteLock();
-
-                internalSubscription = SubscribeToEventStore();
+                await parent.SendAsync(exception);
             }
-            finally
-            {
-                connectionLock.ExitWriteLock();
-            }
-        }
 
-        private EventStoreCatchUpSubscription SubscribeToEventStore()
-        {
-            return connection.SubscribeToStreamFrom(streamName, position, CatchUpSubscriptionSettings.Default, HandleEvent, null, HandleError);
+            await StopAsync();
         }
 
-        private void HandleEvent(EventStoreCatchUpSubscription subscription, ResolvedEvent resolved)
+        protected override async Task OnMessage(IMessage message)
         {
-            if (!CanHandleSubscriptionEvent(subscription))
+            switch (message)
             {
-                return;
-            }
-
-            try
-            {
-                connectionLock.EnterReadLock();
-
-                if (CanHandleSubscriptionEvent(subscription))
-                {
-                    var storedEvent = Formatter.Read(resolved);
-
-                    PublishAsync(storedEvent).Wait();
-
-                    position = resolved.OriginalEventNumber;
-                }
-            }
-            finally
-            {
-                connectionLock.ExitReadLock();
-            }
-        }
+                case SubscribeMessage subscribe when parent == null:
+                    {
+                        parent = subscribe.Parent;
 
-        private void HandleError(EventStoreCatchUpSubscription subscription, SubscriptionDropReason reason, Exception ex)
-        {
-            if (!CanHandleSubscriptionEvent(subscription))
-            {
-                return;
-            }
+                        await CreateProjectionAsync();
 
-            try
-            {
-                connectionLock.EnterUpgradeableReadLock();
+                        break;
+                    }
 
-                if (CanHandleSubscriptionEvent(subscription))
-                {
-                    if (reason == SubscriptionDropReason.ConnectionClosed)
+                case ConnectionFailedMessage connectionFailed when parent != null && subscription == null:
                     {
-                        var utcNow = DateTime.UtcNow;
+                        subscription.Stop();
+                        subscription = null;
 
-                        if (CanReconnect(utcNow))
+                        if (CanReconnect(DateTime.UtcNow))
                         {
-                            RegisterReconnectTime(utcNow);
-
-                            try
-                            {
-                                connectionLock.EnterWriteLock();
-
-                                internalSubscription.Stop();
-                                internalSubscription = null;
-
-                                internalSubscription = SubscribeToEventStore();
-                            }
-                            finally
+                            Task.Delay(ReconnectWaitMs).ContinueWith(t =>
                             {
-                                connectionLock.ExitWriteLock();
-                            }
-
-                            DelayForReconnect().Wait();
+                                SendAsync(new ConnectMessage());
+                            }).Forget();
+                        }
+                        else
+                        {
+                            await SendAsync(connectionFailed.Exception);
+                        }
 
-                            if (!CanHandleSubscriptionEvent(subscription))
-                            {
-                                return;
-                            }
+                        break;
+                    }
 
-                            try
-                            {
-                                connectionLock.EnterWriteLock();
-
-                                if (CanHandleSubscriptionEvent(subscription))
-                                {
-                                    internalSubscription = SubscribeToEventStore();
-                                }
-                            }
-                            finally
-                            {
-                                connectionLock.ExitWriteLock();
-                            }
+                case ConnectMessage connect when parent != null && subscription == null:
+                    {
+                        subscription = connection.SubscribeToStreamFrom(streamName, position, CatchUpSubscriptionSettings.Default, HandleEvent, null, HandleError);
 
-                            return;
-                        }
+                        break;
                     }
 
-                    if (reason != SubscriptionDropReason.UserInitiated)
+                case ReceiveESEventMessage receiveEvent when receiveEvent.Subscription == subscription && parent != null:
                     {
-                        var exception = ex ?? new ConnectionClosedException($"Subscription closed with reason {reason}.");
+                        var storedEvent = Formatter.Read(receiveEvent.Event);
 
-                        publishError?.Invoke(exception);
+                        await parent.SendAsync(new ReceiveEventMessage { Event = storedEvent });
+
+                        position = receiveEvent.Event.OriginalEventNumber;
+
+                        break;
                     }
-                }
-            }
-            finally
-            {
-                connectionLock.ExitUpgradeableReadLock();
             }
         }
 
-        private bool CanHandleSubscriptionEvent(EventStoreCatchUpSubscription subscription)
+        private void HandleEvent(EventStoreCatchUpSubscription s, ResolvedEvent resolved)
         {
-            return !disposeToken.IsCancellationRequested && subscription == internalSubscription;
+            SendAsync(new ReceiveESEventMessage { Event = resolved, Subscription = s }).Forget();
         }
 
-        private bool CanReconnect(DateTime utcNow)
+        private void HandleError(EventStoreCatchUpSubscription s, SubscriptionDropReason reason, Exception ex)
         {
-            return reconnectTimes.Count < ReconnectWindowMax && (reconnectTimes.Count == 0 || (utcNow - reconnectTimes.Peek()) > TimeBetweenReconnects);
-        }
+            if (reason == SubscriptionDropReason.ConnectionClosed)
+            {
+                SendAsync(new ConnectionFailedMessage { Exception = ex });
+            }
+            else if (reason != SubscriptionDropReason.UserInitiated)
+            {
+                var exception = ex ?? new ConnectionClosedException($"Subscription closed with reason {reason}.");
 
-        private async Task PublishAsync(StoredEvent storedEvent)
-        {
-            await publishNext(storedEvent).ConfigureAwait(false);
+                SendAsync(ex).Forget();
+            }
         }
 
         private static long? ParsePosition(string position)
@@ -221,7 +161,7 @@ namespace Squidex.Infrastructure.CQRS.Events
             return long.TryParse(position, out var parsedPosition) ? (long?)parsedPosition : null;
         }
 
-        private void RegisterReconnectTime(DateTime utcNow)
+        private bool CanReconnect(DateTime utcNow)
         {
             reconnectTimes.Enqueue(utcNow);
 
@@ -229,18 +169,8 @@ namespace Squidex.Infrastructure.CQRS.Events
             {
                 reconnectTimes.Dequeue();
             }
-        }
 
-        private async Task DelayForReconnect()
-        {
-            try
-            {
-                await Task.Delay(ReconnectWaitMs, disposeToken.Token).ConfigureAwait(false);
-            }
-            catch (TaskCanceledException)
-            {
-                // Just ignore.
-            }
+            return reconnectTimes.Count < ReconnectWindowMax && (reconnectTimes.Count == 0 || (utcNow - reconnectTimes.Peek()) > TimeBetweenReconnects);
         }
 
         private async Task CreateProjectionAsync()
@@ -263,9 +193,9 @@ namespace Squidex.Infrastructure.CQRS.Events
                 {
                     await projectsManager.CreateContinuousAsync($"${streamName}", projectionConfig, connection.Settings.DefaultUserCredentials);
                 }
-                catch (ProjectionCommandConflictException)
+                catch (Exception ex) when (!(ex is ProjectionCommandConflictException))
                 {
-                    // Projection already exists.
+                    throw;
                 }
             }
         }

src/Squidex.Infrastructure.MongoDb/CQRS/Events/PollingSubscription.cs

@@ -7,21 +7,32 @@
 // ==========================================================================
 
 using System;
+using System.Threading;
 using System.Threading.Tasks;
+using Squidex.Infrastructure.Actors;
+using Squidex.Infrastructure.CQRS.Events.Actors.Messages;
 using Squidex.Infrastructure.Tasks;
 using Squidex.Infrastructure.Timers;
 
+#pragma warning disable SA1401 // Fields must be private
+
 namespace Squidex.Infrastructure.CQRS.Events
 {
-    public sealed class PollingSubscription : DisposableObjectBase, IEventSubscription
+    public sealed class PollingSubscription : Actor, IEventSubscription
     {
         private readonly IEventNotifier eventNotifier;
         private readonly MongoEventStore eventStore;
         private readonly string streamFilter;
+        private CancellationTokenSource ct;
+        private Timer pollTimer;
         private string position;
         private bool isStopped;
-        private IDisposable subscription;
-        private CompletionTimer timer;
+        private IDisposable pollSubscription;
+        private IActor parent;
+
+        private sealed class PollMessage : IMessage
+        {
+        }
 
         public PollingSubscription(MongoEventStore eventStore, IEventNotifier eventNotifier, string streamFilter, string position)
         {
@@ -31,59 +42,82 @@ namespace Squidex.Infrastructure.CQRS.Events
             this.streamFilter = streamFilter;
         }
 
-        protected override void DisposeObject(bool disposing)
+        protected override Task OnStop()
         {
-            if (disposing)
-            {
-                isStopped = true;
+            ct?.Cancel();
 
-                subscription?.Dispose();
+            pollTimer?.Dispose();
+            pollSubscription?.Dispose();
 
-                timer?.StopAsync().Forget();
-            }
+            parent = null;
+
+            return TaskHelper.Done;
         }
 
-        public Task SubscribeAsync(Func<StoredEvent, Task> onNext, Func<Exception, Task> onError = null)
+        protected override async Task OnError(Exception exception)
         {
-            Guard.NotNull(onNext, nameof(onNext));
-
-            if (timer != null)
+            if (parent != null)
             {
-                throw new InvalidOperationException("An handler has already been registered.");
+                await parent.SendAsync(exception);
             }
 
-            timer = new CompletionTimer(5000, async ct =>
+            await StopAsync();
+        }
+
+        protected override async Task OnMessage(IMessage message)
+        {
+            switch (message)
             {
-                try
-                {
-                    await eventStore.GetEventsAsync(async storedEvent =>
+                case SubscribeMessage subscribe when parent == null:
                     {
-                        if (!isStopped)
+                        parent = subscribe.Parent;
+
+                        pollSubscription = eventNotifier.Subscribe(() =>
                         {
-                            await onNext(storedEvent);
-
-                            position = storedEvent.EventPosition;
-                        }
-                    }, ct, streamFilter, position);
-                }
-                catch (Exception ex) when (!(ex is OperationCanceledException))
-                {
-                    if (!isStopped)
+                            SendAsync(new PollMessage()).Forget();
+                        });
+
+                        pollTimer = new Timer(d =>
+                        {
+                            SendAsync(new PollMessage()).Forget();
+                        });
+
+                        pollTimer.Change(0, 5000);
+
+                        break;
+                    }
+
+                case PollMessage poll when parent != null:
                     {
-                        onError?.Invoke(ex);
+                        ct?.Cancel();
+                        ct = new CancellationTokenSource();
+
+                        PollAsync().Forget();
+
+                        break;
                     }
-                }
-            });
 
-            subscription = eventNotifier.Subscribe(() =>
-            {
-                if (!isStopped)
-                {
-                    timer.SkipCurrentDelay();
-                }
-            });
+                case ReceiveEventMessage receiveEvent when parent != null:
+                    {
+                        await parent.SendAsync(receiveEvent);
 
-            return TaskHelper.Done;
+                        position = receiveEvent.Event.EventPosition;
+
+                        break;
+                    }
+            }
+        }
+
+        private async Task PollAsync()
+        {
+            try
+            {
+                await eventStore.GetEventsAsync(e => SendAsync(new ReceiveEventMessage { Event = e }), ct.Token, streamFilter, position);
+            }
+            catch (Exception ex) when (!(ex is OperationCanceledException))
+            {
+                await SendAsync(ex);
+            }
         }
     }
 }

src/Squidex.Infrastructure/Actors/Actor.cs

@@ -0,0 +1,110 @@
+// ==========================================================================
+//  Actor.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using System.Threading.Tasks.Dataflow;
+using Squidex.Infrastructure.Tasks;
+
+#pragma warning disable SA1401 // Fields must be private
+
+namespace Squidex.Infrastructure.Actors
+{
+    public abstract class Actor : IActor, IDisposable
+    {
+        private readonly ActionBlock<IMessage> block;
+        private readonly ReaderWriterLockSlim slimLock = new ReaderWriterLockSlim();
+        private volatile bool isStopped;
+
+        private sealed class StopMessage : IMessage
+        {
+        }
+
+        private sealed class ErrorMessage : IMessage
+        {
+            public Exception Exception;
+        }
+
+        protected Actor()
+        {
+            block = new ActionBlock<IMessage>(Handle, new ExecutionDataflowBlockOptions { BoundedCapacity = 100 });
+        }
+
+        public void Dispose()
+        {
+            StopAsync().Wait();
+        }
+
+        public async Task StopAsync()
+        {
+            isStopped = true;
+
+            await block.SendAsync(new StopMessage());
+
+            await block.Completion;
+        }
+
+        public Task SendAsync(IMessage message)
+        {
+            Guard.NotNull(message, nameof(message));
+
+            return block.SendAsync(message);
+        }
+
+        public Task SendAsync(Exception exception)
+        {
+            Guard.NotNull(exception, nameof(exception));
+
+            return block.SendAsync(new ErrorMessage { Exception = exception });
+        }
+
+        protected virtual Task OnStop()
+        {
+            return TaskHelper.Done;
+        }
+
+        protected virtual Task OnError(Exception exception)
+        {
+            return TaskHelper.Done;
+        }
+
+        protected virtual Task OnMessage(IMessage message)
+        {
+            return TaskHelper.Done;
+        }
+
+        private async Task Handle(IMessage message)
+        {
+            try
+            {
+                if (message is StopMessage)
+                {
+                    block.Complete();
+
+                    await OnStop();
+                }
+                else if (message is ErrorMessage errorMessage)
+                {
+                    await OnError(errorMessage.Exception);
+                }
+                else
+                {
+                    await OnMessage(message);
+                }
+            }
+            catch (Exception ex)
+            {
+                if (!(message is ErrorMessage))
+                {
+                    await block.SendAsync(new ErrorMessage { Exception = ex });
+                }
+            }
+        }
+    }
+}

src/Squidex.Infrastructure/Actors/IActor.cs

@@ -0,0 +1,14 @@
+using System;
+using System.Threading.Tasks;
+
+namespace Squidex.Infrastructure.Actors
+{
+    public interface IActor
+    {
+        Task SendAsync(IMessage message);
+
+        Task SendAsync(Exception exception);
+
+        Task StopAsync();
+    }
+}

src/Squidex.Infrastructure/Actors/IMessage.cs

@@ -0,0 +1,14 @@
+// ==========================================================================
+//  IMessage.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+namespace Squidex.Infrastructure.Actors
+{
+    public interface IMessage
+    {
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/Actors/EventConsumerActor.cs

@@ -0,0 +1,215 @@
+// ==========================================================================
+//  EventReceiver.cs
+//  Squidex Headless CMS
+// ==========================================================================
+//  Copyright (c) Squidex Group
+//  All rights reserved.
+// ==========================================================================
+
+using System;
+using System.Threading.Tasks;
+using Squidex.Infrastructure.Actors;
+using Squidex.Infrastructure.CQRS.Events.Actors.Messages;
+using Squidex.Infrastructure.Log;
+using Squidex.Infrastructure.Tasks;
+
+namespace Squidex.Infrastructure.CQRS.Events.Receivers
+{
+    public sealed class EventConsumerActor : Actor
+    {
+        private readonly EventDataFormatter formatter;
+        private readonly IEventStore eventStore;
+        private readonly IEventConsumerInfoRepository eventConsumerInfoRepository;
+        private readonly ISemanticLog log;
+        private IEventSubscription eventSubscription;
+        private IEventConsumer eventConsumer;
+        private string position;
+
+        public EventConsumerActor(
+            EventDataFormatter formatter,
+            IEventStore eventStore,
+            IEventConsumerInfoRepository eventConsumerInfoRepository,
+            ISemanticLog log)
+        {
+            Guard.NotNull(log, nameof(log));
+            Guard.NotNull(formatter, nameof(formatter));
+            Guard.NotNull(eventStore, nameof(eventStore));
+            Guard.NotNull(eventConsumerInfoRepository, nameof(eventConsumerInfoRepository));
+
+            this.log = log;
+
+            this.formatter = formatter;
+            this.eventStore = eventStore;
+            this.eventConsumerInfoRepository = eventConsumerInfoRepository;
+        }
+
+        public void Subscribe(IEventConsumer eventConsumer)
+        {
+            Guard.NotNull(eventConsumer, nameof(eventConsumer));
+
+            this.eventConsumer = eventConsumer;
+        }
+
+        protected override async Task OnStop()
+        {
+            if (eventSubscription != null)
+            {
+                await eventSubscription.StopAsync();
+            }
+        }
+
+        protected override Task OnError(Exception exception)
+        {
+            return StopAsync(exception);
+        }
+
+        protected override async Task OnMessage(IMessage message)
+        {
+            switch (message)
+            {
+                case StopReceiverMessage stopReceiver:
+                    {
+                        await StopAsync(stopReceiver.Exception);
+
+                        break;
+                    }
+
+                case StartReceiverMessage startReceiver:
+                    {
+                        await StartAsync();
+
+                        break;
+                    }
+
+                case ResetReceiverMessage resetReceiver:
+                    {
+                        await StopAsync();
+                        await ResetAsync();
+                        await StartAsync();
+
+                        break;
+                    }
+
+                case ReceiveEventMessage receiveEvent:
+                    {
+                        await DispatchConsumerAsync(ParseEvent(receiveEvent.Event));
+
+                        break;
+                    }
+            }
+        }
+
+        private async Task StartAsync()
+        {
+            await eventConsumerInfoRepository.CreateAsync(eventConsumer.Name);
+
+            position = (await eventConsumerInfoRepository.FindAsync(eventConsumer.Name)).Position;
+
+            eventSubscription = eventStore.CreateSubscription(eventConsumer.EventsFilter, position);
+            eventSubscription.SendAsync(new SubscribeMessage { Parent = this }).Forget();
+        }
+
+        private async Task StopAsync(Exception exception = null)
+        {
+            if (eventSubscription != null)
+            {
+                await eventSubscription.StopAsync();
+            }
+
+            await eventConsumerInfoRepository.StopAsync(eventConsumer.Name, exception?.Message);
+        }
+
+        private async Task ResetAsync()
+        {
+            var actionId = Guid.NewGuid().ToString();
+            try
+            {
+                log.LogInformation(w => w
+                    .WriteProperty("action", "EventConsumerReset")
+                    .WriteProperty("actionId", actionId)
+                    .WriteProperty("state", "Started")
+                    .WriteProperty("eventConsumer", eventConsumer.Name));
+
+                await eventConsumer.ClearAsync();
+                await eventConsumerInfoRepository.SetPositionAsync(eventConsumer.Name, null, true);
+
+                log.LogInformation(w => w
+                    .WriteProperty("action", "EventConsumerReset")
+                    .WriteProperty("actionId", actionId)
+                    .WriteProperty("state", "Completed")
+                    .WriteProperty("eventConsumer", eventConsumer.Name));
+            }
+            catch (Exception ex)
+            {
+                log.LogFatal(ex, w => w
+                    .WriteProperty("action", "EventConsumerReset")
+                    .WriteProperty("actionId", actionId)
+                    .WriteProperty("state", "Completed")
+                    .WriteProperty("eventConsumer", eventConsumer.GetType().Name));
+
+                throw;
+            }
+        }
+
+        private async Task DispatchConsumerAsync(Envelope<IEvent> @event)
+        {
+            var eventId = @event.Headers.EventId().ToString();
+            var eventType = @event.Payload.GetType().Name;
+            try
+            {
+                log.LogInformation(w => w
+                    .WriteProperty("action", "HandleEvent")
+                    .WriteProperty("actionId", eventId)
+                    .WriteProperty("state", "Started")
+                    .WriteProperty("eventId", eventId)
+                    .WriteProperty("eventType", eventType)
+                    .WriteProperty("eventConsumer", eventConsumer.Name));
+
+                await eventConsumer.On(@event);
+
+                log.LogInformation(w => w
+                    .WriteProperty("action", "HandleEvent")
+                    .WriteProperty("actionId", eventId)
+                    .WriteProperty("state", "Completed")
+                    .WriteProperty("eventId", eventId)
+                    .WriteProperty("eventType", eventType)
+                    .WriteProperty("eventConsumer", eventConsumer.Name));
+            }
+            catch (Exception ex)
+            {
+                log.LogError(ex, w => w
+                    .WriteProperty("action", "HandleEvent")
+                    .WriteProperty("actionId", eventId)
+                    .WriteProperty("state", "Started")
+                    .WriteProperty("eventId", eventId)
+                    .WriteProperty("eventType", eventType)
+                    .WriteProperty("eventConsumer", eventConsumer.Name));
+
+                throw;
+            }
+        }
+
+        private Envelope<IEvent> ParseEvent(StoredEvent message)
+        {
+            try
+            {
+                var @event = formatter.Parse(message.Data);
+
+                @event.SetEventPosition(message.EventPosition);
+                @event.SetEventStreamNumber(message.EventStreamNumber);
+
+                return @event;
+            }
+            catch (Exception ex)
+            {
+                log.LogFatal(ex, w => w
+                    .WriteProperty("action", "ParseEvent")
+                    .WriteProperty("state", "Failed")
+                    .WriteProperty("eventId", message.Data.EventId.ToString())
+                    .WriteProperty("eventPosition", message.EventPosition));
+
+                throw;
+            }
+        }
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/Actors/Messages/ReceiveEventMessage.cs

@@ -0,0 +1,9 @@
+using Squidex.Infrastructure.Actors;
+
+namespace Squidex.Infrastructure.CQRS.Events.Actors.Messages
+{
+    public sealed class ReceiveEventMessage : IMessage
+    {
+        public StoredEvent Event { get; set; }
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/Actors/Messages/ResetReceiverMessage.cs

@@ -0,0 +1,8 @@
+using Squidex.Infrastructure.Actors;
+
+namespace Squidex.Infrastructure.CQRS.Events.Actors.Messages
+{
+    public sealed class ResetReceiverMessage : IMessage
+    {
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/Actors/Messages/StartReceiverMessage.cs

@@ -0,0 +1,8 @@
+using Squidex.Infrastructure.Actors;
+
+namespace Squidex.Infrastructure.CQRS.Events.Actors.Messages
+{
+    public sealed class StartReceiverMessage : IMessage
+    {
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/Actors/Messages/StopReceiverMessage.cs

@@ -0,0 +1,10 @@
+using System;
+using Squidex.Infrastructure.Actors;
+
+namespace Squidex.Infrastructure.CQRS.Events.Actors.Messages
+{
+    public sealed class StopReceiverMessage : IMessage
+    {
+        public Exception Exception { get; set; }
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/Actors/Messages/SubscribeMessage.cs

@@ -0,0 +1,9 @@
+using Squidex.Infrastructure.Actors;
+
+namespace Squidex.Infrastructure.CQRS.Events.Actors.Messages
+{
+    public sealed class SubscribeMessage : IMessage
+    {
+        public IActor Parent { get; set; }
+    }
+}

src/Squidex.Infrastructure/CQRS/Events/IEventSubscription.cs

@@ -6,13 +6,11 @@
 //  All rights reserved.
 // ==========================================================================
 
-using System;
-using System.Threading.Tasks;
+using Squidex.Infrastructure.Actors;
 
 namespace Squidex.Infrastructure.CQRS.Events
 {
-    public interface IEventSubscription : IDisposable
+    public interface IEventSubscription : IActor
     {
-        Task SubscribeAsync(Func<StoredEvent, Task> onNext, Func<Exception, Task> onError = null);
     }
 }

src/Squidex.Infrastructure/Squidex.Infrastructure.csproj

@@ -11,6 +11,7 @@
     <PackageReference Include="ImageSharp" Version="1.0.0-alpha9-00191" />
     <PackageReference Include="Microsoft.Extensions.Caching.Abstractions" Version="1.1.2" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="1.1.2" />
+    <PackageReference Include="Microsoft.Tpl.Dataflow" Version="4.5.24" />
     <PackageReference Include="Newtonsoft.Json" Version="10.0.3" />
     <PackageReference Include="NodaTime" Version="2.2.0" />
     <PackageReference Include="RefactoringEssentials" Version="5.2.0" />