Delegate HasBacklog to ConsumerRepository

disruptor-net · Aug 31, 2019 · 99d2934 · 99d2934
1 parent 38f50b0
commit 99d2934
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 12 deletions.
diff --git a/src/Disruptor/Dsl/ConsumerRepository.cs b/src/Disruptor/Dsl/ConsumerRepository.cs
@@ -43,6 +43,25 @@ public void Add<T>(WorkerPool<T> workerPool, ISequenceBarrier sequenceBarrier)
             }
         }
 
+        public bool HasBacklog(long cursor, bool includeStopped)
+        {
+            foreach (var consumerInfo in _consumerInfos)
+            {
+                if ((includeStopped || consumerInfo.IsRunning) && consumerInfo.IsEndOfChain)
+                {
+                    foreach (var sequence in consumerInfo.Sequences)
+                    {
+                        if (cursor > sequence.Value)
+                        {
+                            return true;
+                        }
+                    }
+                }
+            }
+
+            return false;
+        }
+
         public ISequence[] GetLastSequenceInChain(bool includeStopped)
         {
             var lastSequence = new List<ISequence>();

diff --git a/src/Disruptor/Dsl/Disruptor.cs b/src/Disruptor/Dsl/Disruptor.cs
@@ -396,12 +396,7 @@ public void Shutdown(TimeSpan timeout)
         private bool HasBacklog()
         {
             var cursor = _ringBuffer.Cursor;
-            foreach (var sequence in _consumerRepository.GetLastSequenceInChain(false))
-            {
-                if (cursor > sequence.Value)
-                    return true;
-            }
-            return false;
+            return _consumerRepository.HasBacklog(cursor, false);
         }
 
         internal EventHandlerGroup<T> CreateEventProcessors(ISequence[] barrierSequences, IEventHandler<T>[] eventHandlers)

diff --git a/src/Disruptor/Dsl/ValueDisruptor.cs b/src/Disruptor/Dsl/ValueDisruptor.cs
@@ -314,12 +314,7 @@ public void Shutdown(TimeSpan timeout)
         private bool HasBacklog()
         {
             var cursor = _ringBuffer.Cursor;
-            foreach (var sequence in _consumerRepository.GetLastSequenceInChain(false))
-            {
-                if (cursor > sequence.Value)
-                    return true;
-            }
-            return false;
+            return _consumerRepository.HasBacklog(cursor, false);
         }
 
         internal ValueEventHandlerGroup<T> CreateEventProcessors(ISequence[] barrierSequences, IValueEventHandler<T>[] eventHandlers)