Merge pull request #715 from dolittle/plug

dolittle · Oct 5, 2022 · 278f839 · 278f839
2 parents 131baa0 + cd819e2
commit 278f839
Showing 1 changed file with 9 additions and 2 deletions.
diff --git a/Source/Events.Processing/Streams/AbstractScopedStreamProcessor.cs b/Source/Events.Processing/Streams/AbstractScopedStreamProcessor.cs
@@ -28,8 +28,8 @@ public abstract class AbstractScopedStreamProcessor
     readonly IEventFetcherPolicies _eventFetcherPolicies;
     readonly IStreamEventWatcher _eventWaiter;
     readonly object _setPositionLock = new();
-    CancellationTokenSource _resetStreamProcessor;
-    TaskCompletionSource<Try<StreamPosition>> _resetStreamProcessorCompletionSource;
+    CancellationTokenSource? _resetStreamProcessor;
+    TaskCompletionSource<Try<StreamPosition>>? _resetStreamProcessorCompletionSource;
     Func<TenantId, CancellationToken, Task<Try>> _resetStreamProcessorAction;
     IStreamProcessorState _currentState;
     bool _started;
@@ -344,6 +344,7 @@ async Task BeginProcessing(CancellationToken cancellationToken)
                         _resetStreamProcessorCompletionSource = default;
                         _resetStreamProcessorAction = default;
                     }
+
                     try
                     {
                         _currentState = await Catchup(_currentState, _resetStreamProcessor.Token).ConfigureAwait(false);
@@ -362,6 +363,11 @@ await _eventWaiter.WaitForEvent(
                     {
                         tryGetEvents = ex;
                     }
+                    finally
+                    {
+                        _resetStreamProcessor?.Dispose();
+                        _resetStreamProcessor = null;
+                    }
                 }
                 while (!tryGetEvents.Success && !cancellationToken.IsCancellationRequested);
 
@@ -385,6 +391,7 @@ await _eventWaiter.WaitForEvent(
         finally
         {
             _resetStreamProcessor?.Dispose();
+            _resetStreamProcessor = null;
         }
     }
 }