dorkbox
/
MessageBus
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Correct check for more messages in LMAX

This commit is contained in:
Robinson 2021-04-09 20:19:23 +02:00
parent f3549ec7d6
commit 320b7ea738
1 changed files with 5 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
src/dorkbox/messageBus/publication/LmaxDisruptor.java
View File

@ -31,6 +31,7 @@ import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.Sequencer; import com.lmax.disruptor.Sequencer;
import com.lmax.disruptor.WaitStrategy; import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkProcessor; import com.lmax.disruptor.WorkProcessor;
import com.lmax.disruptor.util.Util;
import dorkbox.messageBus.error.ErrorHandler; import dorkbox.messageBus.error.ErrorHandler;
import dorkbox.messageBus.publication.disruptor.EventBusFactory; import dorkbox.messageBus.publication.disruptor.EventBusFactory;
@ -303,20 +304,14 @@ class LmaxDisruptor implements Publisher {
ringBuffer.publish(seq); ringBuffer.publish(seq);
} }
@Override
public public
boolean hasPendingMessages() { boolean hasPendingMessages() {
// from workerPool.drainAndHalt() // modified from workerPool.drainAndHalt()
Sequence[] workerSequences = getSequences(); return ringBuffer.getCursor() > Util.getMinimumSequence(getSequences());
final long cursor = ringBuffer.getCursor();
for (Sequence s : workerSequences) {
if (cursor > s.get()) {
return true;
}
}
return false;
} }
@Override
public public
void shutdown() { void shutdown() {
// let some messages finish publishing // let some messages finish publishing