package rxbroadcast;

import rxbroadcast.time.Clock;
import rxbroadcast.time.LamportClock;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.function.Consumer;

public final class SingleSourceFifoOrder<T> implements BroadcastOrder<Timestamped<T>, T> {
    @SuppressWarnings("WeakerAccess")
    public static final boolean DROP_LATE = true;

    private final Clock clock = new LamportClock();

    private final Map<Sender, SortedSet<Timestamped<T>>> pendingSets;

    private final Map<Sender, Long> expectedTimestamps;

    private final boolean dropLateMessages;

    public SingleSourceFifoOrder() {
        this(!DROP_LATE);
    }

    @SuppressWarnings("WeakerAccess")
    public SingleSourceFifoOrder(final boolean dropLateMessages) {
        this.pendingSets = new HashMap<>();
        this.expectedTimestamps = new HashMap<>();
        this.dropLateMessages = dropLateMessages;
    }

    @Override
    public Timestamped<T> prepare(final T value) {
        return clock.tick(time -> new Timestamped<>(time, value));
    }

    @Override
    public void receive(final Sender sender, final Consumer<T> consumer, final Timestamped<T> value) {
        expectedTimestamps.putIfAbsent(sender, 1L);

        if (dropLateMessages) {
            if (Long.compareUnsigned(value.timestamp, expectedTimestamps.get(sender)) >= 0) {
                consumer.accept(value.value);
                expectedTimestamps.compute(sender, (k, v) -> value.timestamp + 1);
            }

            return;
        }

        final SortedSet<Timestamped<T>> pendingSet = pendingSets.computeIfAbsent(sender, k -> new TreeSet<>());

        pendingSet.add(value);

        final Iterator<Timestamped<T>> iterator = pendingSet.iterator();
        while (iterator.hasNext()) {
            final Timestamped<T> tv = iterator.next();
            if (tv.timestamp < expectedTimestamps.get(sender)) {
                iterator.remove();
                continue;
            }
            if (tv.timestamp > expectedTimestamps.get(sender)) {
                break;
            }

            consumer.accept(tv.value);
            expectedTimestamps.compute(sender, (k, v) -> tv.timestamp + 1);
            iterator.remove();
        }
    }

    int queueSize() {
        return pendingSets.values().stream().mapToInt(SortedSet::size).sum();
    }
}

@@ -350,90 +350,235 @@
     
-@SuppressWarnings(%22WeakerAccess%22)%0A    public static final boolean DROP_LATE = true
+public enum SingleSourceFifoOrderQueueOption %7B%0A        DROP,%0A        QUEUE,%0A    %7D%0A%0A    @SuppressWarnings(%22WeakerAccess%22)%0A    public static final SingleSourceFifoOrderQueueOption DROP_LATE = SingleSourceFifoOrderQueueOption.DROP
 ;%0A%0A 
@@ -856,17 +856,45 @@
 his(
-!DROP_LAT
+SingleSourceFifoOrderQueueOption.QUEU
 E);%0A
@@ -981,32 +981,49 @@
 nal 
-boolean dropLateMessages
+SingleSourceFifoOrderQueueOption queueOpt
 ) %7B%0A
@@ -1149,32 +1149,65 @@
 sages = 
-dropLateMessages
+queueOpt == SingleSourceFifoOrderQueueOption.DROP
 ;%0A    %7D%0A

Replace ssf boolean flag with DROP+QUEUE enum