- event processing

Author: eben-roux

Shuttle.Recall/EventProcessing/EventProcessor.cs

@@ -278,39 +278,32 @@ private async Task<IEventProcessor> StartAsync(bool sync)
                 return this;
             }
 
-            Started = true;
             Asynchronous = !sync;
 
             foreach (var projectionAggregation in _projectionAggregations.Values)
             {
                 projectionAggregation.AddEventTypes();
             }
 
-            try
+            var startupPipeline = _pipelineFactory.GetPipeline<EventProcessorStartupPipeline>();
+
+            if (sync)
+            {
+                startupPipeline.Execute();
+            }
+            else
             {
-                var startupPipeline = _pipelineFactory.GetPipeline<EventProcessorStartupPipeline>();
+                await startupPipeline.ExecuteAsync().ConfigureAwait(false);
+            }
 
-                if (sync)
-                {
-                    startupPipeline.Execute();
-                }
-                else
-                {
-                    await startupPipeline.ExecuteAsync().ConfigureAwait(false);
-                }
+            _cancellationTokenSource = new CancellationTokenSource();
+            _cancellationToken = _cancellationTokenSource.Token;
 
-                _cancellationTokenSource = new CancellationTokenSource();
-                _cancellationToken = _cancellationTokenSource.Token;
+            _eventProcessorThreadPool = startupPipeline.State.Get<IProcessorThreadPool>("EventProcessorThreadPool");
 
-                _eventProcessorThreadPool = startupPipeline.State.Get<IProcessorThreadPool>("EventProcessorThreadPool");
+            _sequenceNumberTailThread.Start();
 
-                _sequenceNumberTailThread.Start();
-            }
-            catch
-            {
-                Started = false;
-                throw;
-            }
+            Started = true;
 
             return this;
         }

Shuttle.Recall/Pipeline/Events.cs

@@ -170,15 +170,15 @@ public class OnAfterEncryptEvent : PipelineEvent
     {
     }
 
-    public class OnBeforeStartEventProcessingEvent : PipelineEvent
+    public class OnBeforeStartEventProcessing : PipelineEvent
     {
     }
 
-    public class OnStartEventProcessingEvent : PipelineEvent
+    public class OnStartEventProcessing : PipelineEvent
     {
     }
 
-    public class OnAfterStartEventProcessingEvent : PipelineEvent
+    public class OnAfterStartEventProcessing : PipelineEvent
     {
     }
 

Shuttle.Recall/Pipeline/Observers/EventProjection/StartupEventProcessingObserver.cs

@@ -24,12 +24,12 @@ public StartupEventProcessingObserver(IOptions<EventStoreOptions> eventStoreOpti
             _eventStoreConfiguration = Guard.AgainstNull(eventStoreConfiguration, nameof(eventStoreConfiguration));
         }
 
-        public void Execute(OnStartEventProcessingEvent pipelineEvent)
+        public void Execute(OnStartEventProcessing pipelineEvent)
         {
             ExecuteAsync(pipelineEvent).GetAwaiter().GetResult();
         }
 
-        public async Task ExecuteAsync(OnStartEventProcessingEvent pipelineEvent)
+        public async Task ExecuteAsync(OnStartEventProcessing pipelineEvent)
         {
             Guard.AgainstNull(pipelineEvent, nameof(pipelineEvent));
 
@@ -91,7 +91,7 @@ private async Task ExecuteAsync(OnStartThreadPools pipelineEvent, bool sync)
     }
 
     public interface IStartupEventProcessingObserver : 
-        IPipelineObserver<OnStartEventProcessingEvent>,
+        IPipelineObserver<OnStartEventProcessing>,
         IPipelineObserver<OnConfigureThreadPools>,
         IPipelineObserver<OnStartThreadPools>
     {

Shuttle.Recall/Pipeline/Pipelines/EventProcessorStartupPipeline.cs

@@ -13,9 +13,9 @@ public EventProcessorStartupPipeline(IStartupEventProcessingObserver startupEven
             RegisterStage("Process")
                 .WithEvent<OnStartTransactionScope>()
                 .WithEvent<OnAfterStartTransactionScope>()
-                .WithEvent<OnBeforeStartEventProcessingEvent>()
-                .WithEvent<OnStartEventProcessingEvent>()
-                .WithEvent<OnAfterStartEventProcessingEvent>()
+                .WithEvent<OnBeforeStartEventProcessing>()
+                .WithEvent<OnStartEventProcessing>()
+                .WithEvent<OnAfterStartEventProcessing>()
                 .WithEvent<OnCompleteTransactionScope>()
                 .WithEvent<OnDisposeTransactionScope>()
                 .WithEvent<OnConfigureThreadPools>()