SequenceBuffer.java

import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.function.Consumer;

public class SequenceBuffer {

    private static final long GAP_TIMEOUT_MS = 100; // skip gap after 100ms

    private final TreeMap<Integer, Order> buffer       = new TreeMap<>();
    private final BlockingQueue<Order>    outputQueue;
    private int  nextExpected  = 1;
    private long gapDetectedAt = -1;

    public SequenceBuffer(BlockingQueue<Order> outputQueue) {
        this.outputQueue = outputQueue;
    }

    public synchronized void add(Order order) throws InterruptedException {
        if (order.getSequenceId() < nextExpected) {
            // Duplicate or very late order — discard
            return;
        }
        buffer.put(order.getSequenceId(), order);
        flush();
    }

    private void flush() throws InterruptedException {
        while (!buffer.isEmpty()) {
            int firstKey = buffer.firstKey();

            if (firstKey == nextExpected) {
                // In-order: release it
                outputQueue.put(buffer.remove(firstKey));
                nextExpected++;
                gapDetectedAt = -1; // reset gap timer

            } else {
                // Gap detected — start or check timeout
                long now = System.currentTimeMillis();
                if (gapDetectedAt == -1) {
                    gapDetectedAt = now; // start the clock
                } else if (now - gapDetectedAt > GAP_TIMEOUT_MS) {
                    // Timeout: skip the missing sequence ID
                    System.out.printf("[SeqBuffer] Skipping missing seqId=%d after timeout%n",
                        nextExpected);
                    nextExpected++;
                    gapDetectedAt = -1;
                    // Loop again to try the next one
                    continue;
                }
                break; // wait for the missing order
            }
        }
    }
}