유한 버퍼 계약

import java.util.ArrayDeque;
import java.util.Queue;

public final class SilentDropBuffer {
    private final Queue<String> queue = new ArrayDeque<>();
    private final int capacity;

    public SilentDropBuffer(int capacity) {
        this.capacity = capacity;
    }

    public synchronized void put(String value) {
        if (queue.size() == capacity) {
            return;
        }
        queue.add(value);
    }

    public synchronized String take() {
        return queue.poll();
    }

    public static void main(String[] args) {
        SilentDropBuffer buffer = new SilentDropBuffer(2);
        for (String event : new String[]{"login", "search", "purchase"}) {
            buffer.put(event);
            System.out.println("accepted=" + event);
        }
        System.out.println("saved=" + buffer.take());
        System.out.println("saved=" + buffer.take());
        System.out.println("saved=" + buffer.take());
    }
}

import java.util.ArrayDeque;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Queue;

public final class ObservableBoundedBuffer<E> {
    public enum OfferResult { ACCEPTED, FULL }

    private final Queue<E> elements = new ArrayDeque<>();
    private final int capacity;

    public ObservableBoundedBuffer(int capacity) {
        if (capacity < 1) {
            throw new IllegalArgumentException("capacity=" + capacity);
        }
        this.capacity = capacity;
    }

    public synchronized OfferResult offer(E element) {
        Objects.requireNonNull(element, "element");
        if (elements.size() >= capacity) {
            return OfferResult.FULL;
        }
        elements.add(element);
        return OfferResult.ACCEPTED;
    }

    public synchronized Optional<E> poll() {
        return Optional.ofNullable(elements.poll());
    }

    public synchronized int size() {
        return elements.size();
    }

    public synchronized List<E> snapshot() {
        return List.copyOf(elements);
    }

    public static void main(String[] args) {
        ObservableBoundedBuffer<String> buffer = new ObservableBoundedBuffer<>(2);
        int accepted = 0;
        int rejected = 0;
        for (String event : List.of("login", "search", "purchase")) {
            if (buffer.offer(event) == OfferResult.ACCEPTED) accepted++;
            else rejected++;
        }
        int consumed = 0;
        while (buffer.poll().isPresent()) consumed++;
        System.out.printf("accepted=%d rejected=%d consumed=%d remaining=%d%n",
                accepted, rejected, consumed, buffer.size());
    }
}

import java.util.Optional;

public final class PollingDrainDemo {
    private static final class RunState {
        private volatile boolean productionFinished;
        private int handled;
    }

    public static void main(String[] args) throws InterruptedException {
        ObservableBoundedBuffer<String> buffer = new ObservableBoundedBuffer<>(4);
        RunState state = new RunState();

        Thread producer = Thread.ofPlatform().start(() -> {
            for (String value : new String[]{"A", "B", "C", "D"}) {
                ObservableBoundedBuffer.OfferResult result = buffer.offer(value);
                if (result != ObservableBoundedBuffer.OfferResult.ACCEPTED) {
                    throw new IllegalStateException("unexpected overflow: " + value);
                }
            }
            state.productionFinished = true;
        });

        Thread consumer = Thread.ofPlatform().start(() -> {
            while (!state.productionFinished || buffer.size() > 0) {
                Optional<String> next = buffer.poll();
                if (next.isPresent()) {
                    state.handled++;
                } else {
                    Thread.onSpinWait();
                }
            }
        });

        producer.join();
        consumer.join();
        System.out.println("handled=" + state.handled);
    }
}