queue;
+
+ @Setup(Level.Iteration)
+ public void setUp() {
+ queue = CloseableQueues.bounded(capacity, strategy, Duration.ofNanos(1_000));
+ }
+
+ @TearDown(Level.Iteration)
+ public void tearDown() {
+ queue.closeNow();
+ }
+ }
+
+ @Benchmark
+ @Group("producerConsumer")
+ @GroupThreads(1)
+ public boolean offer(QueueState state) throws InterruptedException {
+ return state.queue.offer(1, OPERATION_TIMEOUT);
+ }
+
+ @Benchmark
+ @Group("producerConsumer")
+ @GroupThreads(1)
+ public int poll(QueueState state) throws InterruptedException {
+ return state.queue.poll(OPERATION_TIMEOUT).orElse(0);
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/batching/AdaptiveBatchSizer.java b/src/main/java/dev/azamir/primitives/batching/AdaptiveBatchSizer.java
new file mode 100644
index 0000000..fc52003
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/batching/AdaptiveBatchSizer.java
@@ -0,0 +1,99 @@
+package dev.azamir.primitives.batching;
+
+import java.util.Objects;
+
+/**
+ * Estimates a safe batch size from a representative serialized sample.
+ *
+ * The calculation separates fixed overhead from per-record payload, projects how many records fit
+ * in the target capacity, applies a safety factor, and clamps the result to configured limits.
+ */
+public final class AdaptiveBatchSizer {
+ private final long targetCapacityBytes;
+ private final double safetyFactor;
+ private final int minimumBatchSize;
+ private final int maximumBatchSize;
+
+ public AdaptiveBatchSizer(long targetCapacityBytes, double safetyFactor) {
+ this(targetCapacityBytes, safetyFactor, 1, Integer.MAX_VALUE);
+ }
+
+ public AdaptiveBatchSizer(
+ long targetCapacityBytes,
+ double safetyFactor,
+ int minimumBatchSize,
+ int maximumBatchSize) {
+ if (targetCapacityBytes <= 0) {
+ throw new IllegalArgumentException("targetCapacityBytes must be greater than zero");
+ }
+ if (!Double.isFinite(safetyFactor) || safetyFactor <= 0 || safetyFactor > 1) {
+ throw new IllegalArgumentException("safetyFactor must be in the range (0, 1]");
+ }
+ if (minimumBatchSize <= 0) {
+ throw new IllegalArgumentException("minimumBatchSize must be greater than zero");
+ }
+ if (maximumBatchSize < minimumBatchSize) {
+ throw new IllegalArgumentException("maximumBatchSize must be at least minimumBatchSize");
+ }
+ this.targetCapacityBytes = targetCapacityBytes;
+ this.safetyFactor = safetyFactor;
+ this.minimumBatchSize = minimumBatchSize;
+ this.maximumBatchSize = maximumBatchSize;
+ }
+
+ public int estimate(
+ int sampleRecordCount, long serializedSampleBytes, long fixedOverheadBytes) {
+ if (sampleRecordCount < 0) {
+ throw new IllegalArgumentException("sampleRecordCount must not be negative");
+ }
+ if (sampleRecordCount == 0) {
+ return 0;
+ }
+ if (serializedSampleBytes < 0 || fixedOverheadBytes < 0) {
+ throw new IllegalArgumentException("serialized sizes must not be negative");
+ }
+ if (serializedSampleBytes < fixedOverheadBytes) {
+ throw new IllegalArgumentException(
+ "serializedSampleBytes must include and therefore be at least fixedOverheadBytes");
+ }
+
+ if (fixedOverheadBytes >= targetCapacityBytes) {
+ return clamp(sampleRecordCount);
+ }
+
+ long payloadBytes = serializedSampleBytes - fixedOverheadBytes;
+ if (payloadBytes == 0) {
+ return clamp(sampleRecordCount);
+ }
+
+ double averageRecordBytes = (double) payloadBytes / sampleRecordCount;
+ double usableCapacity = targetCapacityBytes - fixedOverheadBytes;
+ long projected = (long) Math.floor((usableCapacity / averageRecordBytes) * safetyFactor);
+ if (projected <= 0) {
+ return clamp(sampleRecordCount);
+ }
+ return clamp(projected);
+ }
+
+ public int estimate(
+ int sampleRecordCount,
+ T serializedSample,
+ RetryingSizeProbe totalSizeProbe,
+ long fixedOverheadBytes) {
+ Objects.requireNonNull(totalSizeProbe, "totalSizeProbe");
+ return estimate(
+ sampleRecordCount, totalSizeProbe.measureBytes(serializedSample), fixedOverheadBytes);
+ }
+
+ public long targetCapacityBytes() {
+ return targetCapacityBytes;
+ }
+
+ public double safetyFactor() {
+ return safetyFactor;
+ }
+
+ private int clamp(long value) {
+ return (int) Math.max(minimumBatchSize, Math.min(maximumBatchSize, value));
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/batching/CheckedSizeProbe.java b/src/main/java/dev/azamir/primitives/batching/CheckedSizeProbe.java
new file mode 100644
index 0000000..5c326eb
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/batching/CheckedSizeProbe.java
@@ -0,0 +1,7 @@
+package dev.azamir.primitives.batching;
+
+/** Measures a serialized size and may fail with a checked exception. */
+@FunctionalInterface
+public interface CheckedSizeProbe {
+ long measureBytes(T value) throws Exception;
+}
diff --git a/src/main/java/dev/azamir/primitives/batching/RetryingSizeProbe.java b/src/main/java/dev/azamir/primitives/batching/RetryingSizeProbe.java
new file mode 100644
index 0000000..523930d
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/batching/RetryingSizeProbe.java
@@ -0,0 +1,41 @@
+package dev.azamir.primitives.batching;
+
+import java.util.Objects;
+
+/** Retries transient serialized-size measurement failures up to a fixed attempt budget. */
+public final class RetryingSizeProbe {
+ private final CheckedSizeProbe delegate;
+ private final int maxAttempts;
+
+ public RetryingSizeProbe(CheckedSizeProbe delegate, int maxAttempts) {
+ this.delegate = Objects.requireNonNull(delegate, "delegate");
+ if (maxAttempts <= 0) {
+ throw new IllegalArgumentException("maxAttempts must be greater than zero");
+ }
+ this.maxAttempts = maxAttempts;
+ }
+
+ public long measureBytes(T value) {
+ Exception lastFailure = null;
+ for (int attempt = 1; attempt <= maxAttempts; attempt++) {
+ try {
+ long measured = delegate.measureBytes(value);
+ if (measured < 0) {
+ throw new IllegalStateException("measured size must not be negative");
+ }
+ return measured;
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw new SizeProbeException("size measurement was interrupted", interrupted);
+ } catch (Exception failure) {
+ lastFailure = failure;
+ }
+ }
+ throw new SizeProbeException(
+ "could not measure serialized size after " + maxAttempts + " attempts", lastFailure);
+ }
+
+ public int maxAttempts() {
+ return maxAttempts;
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/batching/SizeProbeException.java b/src/main/java/dev/azamir/primitives/batching/SizeProbeException.java
new file mode 100644
index 0000000..163f609
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/batching/SizeProbeException.java
@@ -0,0 +1,8 @@
+package dev.azamir.primitives.batching;
+
+/** Raised when a serialized size cannot be measured within the configured retry budget. */
+public final class SizeProbeException extends RuntimeException {
+ public SizeProbeException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/BoundedBlockingQueue.java b/src/main/java/dev/azamir/primitives/concurrent/BoundedBlockingQueue.java
new file mode 100644
index 0000000..c9c09f0
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/BoundedBlockingQueue.java
@@ -0,0 +1,321 @@
+package dev.azamir.primitives.concurrent;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A bounded FIFO ring buffer with condition-based blocking and an explicit close lifecycle.
+ *
+ * Closing the queue rejects new elements, wakes blocked producers and consumers, and still allows
+ * consumers to drain elements that were already accepted. Once closed and drained, {@link #take()}
+ * throws {@link QueueClosedException} while polling methods return an empty result.
+ *
+ * @param element type
+ */
+public final class BoundedBlockingQueue implements CloseableBlockingQueue {
+ private final Object[] elements;
+ private final ReentrantLock lock;
+ private final Condition notEmpty;
+ private final Condition notFull;
+
+ private int head;
+ private int tail;
+ private int size;
+ private boolean closed;
+
+ public BoundedBlockingQueue(int capacity) {
+ this(capacity, false);
+ }
+
+ public BoundedBlockingQueue(int capacity, boolean fair) {
+ if (capacity <= 0) {
+ throw new IllegalArgumentException("capacity must be greater than zero");
+ }
+ elements = new Object[capacity];
+ lock = new ReentrantLock(fair);
+ notEmpty = lock.newCondition();
+ notFull = lock.newCondition();
+ }
+
+ @Override
+ public void put(E element) throws InterruptedException {
+ Objects.requireNonNull(element, "element");
+ lock.lockInterruptibly();
+ try {
+ while (size == elements.length && !closed) {
+ notFull.await();
+ }
+ ensureOpenForWrite();
+ enqueue(element);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public boolean offer(E element, Duration timeout) throws InterruptedException {
+ Objects.requireNonNull(element, "element");
+ Objects.requireNonNull(timeout, "timeout");
+ if (timeout.isNegative()) {
+ throw new IllegalArgumentException("timeout must not be negative");
+ }
+
+ long remaining = toNanosSaturated(timeout);
+ lock.lockInterruptibly();
+ try {
+ while (size == elements.length && !closed) {
+ if (remaining <= 0) {
+ return false;
+ }
+ remaining = notFull.awaitNanos(remaining);
+ }
+ ensureOpenForWrite();
+ enqueue(element);
+ return true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ lock.lockInterruptibly();
+ try {
+ while (size == 0 && !closed) {
+ notEmpty.await();
+ }
+ if (size == 0) {
+ throw new QueueClosedException("queue is closed and drained");
+ }
+ return dequeue();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Optional poll() {
+ lock.lock();
+ try {
+ return size == 0 ? Optional.empty() : Optional.of(dequeue());
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Optional poll(Duration timeout) throws InterruptedException {
+ Objects.requireNonNull(timeout, "timeout");
+ if (timeout.isNegative()) {
+ throw new IllegalArgumentException("timeout must not be negative");
+ }
+
+ long remaining = toNanosSaturated(timeout);
+ lock.lockInterruptibly();
+ try {
+ while (size == 0 && !closed) {
+ if (remaining <= 0) {
+ return Optional.empty();
+ }
+ remaining = notEmpty.awaitNanos(remaining);
+ }
+ return size == 0 ? Optional.empty() : Optional.of(dequeue());
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Optional peek() {
+ lock.lock();
+ try {
+ return size == 0 ? Optional.empty() : Optional.of(elementAt(0));
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int drainTo(Collection target, int maxElements) {
+ Objects.requireNonNull(target, "target");
+ if (maxElements < 0) {
+ throw new IllegalArgumentException("maxElements must not be negative");
+ }
+
+ lock.lock();
+ try {
+ int drained = Math.min(size, maxElements);
+ for (int i = 0; i < drained; i++) {
+ target.add(dequeue());
+ }
+ return drained;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int clear() {
+ lock.lock();
+ try {
+ int removed = clearElements();
+ if (removed > 0) {
+ notFull.signalAll();
+ }
+ return removed;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int closeNow() {
+ lock.lock();
+ try {
+ closed = true;
+ int discarded = clearElements();
+ notEmpty.signalAll();
+ notFull.signalAll();
+ return discarded;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public boolean contains(Object value) {
+ return indexOf(value) >= 0;
+ }
+
+ @Override
+ public int indexOf(Object value) {
+ lock.lock();
+ try {
+ for (int i = 0; i < size; i++) {
+ if (Objects.equals(value, elements[(head + i) % elements.length])) {
+ return i;
+ }
+ }
+ return -1;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public List snapshot() {
+ lock.lock();
+ try {
+ List copy = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ copy.add(elementAt(i));
+ }
+ return List.copyOf(copy);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void close() {
+ lock.lock();
+ try {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ notEmpty.signalAll();
+ notFull.signalAll();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public boolean isClosed() {
+ lock.lock();
+ try {
+ return closed;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int size() {
+ lock.lock();
+ try {
+ return size;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int capacity() {
+ return elements.length;
+ }
+
+ @Override
+ public int remainingCapacity() {
+ lock.lock();
+ try {
+ return elements.length - size;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void ensureOpenForWrite() {
+ if (closed) {
+ throw new QueueClosedException("queue is closed for writes");
+ }
+ }
+
+ private void enqueue(E element) {
+ elements[tail] = element;
+ tail = (tail + 1) % elements.length;
+ size++;
+ notEmpty.signal();
+ }
+
+ @SuppressWarnings("unchecked")
+ private E dequeue() {
+ E element = (E) elements[head];
+ elements[head] = null;
+ head = (head + 1) % elements.length;
+ size--;
+ notFull.signal();
+ return element;
+ }
+
+ private int clearElements() {
+ int removed = size;
+ for (int i = 0; i < removed; i++) {
+ elements[(head + i) % elements.length] = null;
+ }
+ head = 0;
+ tail = 0;
+ size = 0;
+ return removed;
+ }
+
+ @SuppressWarnings("unchecked")
+ private E elementAt(int logicalIndex) {
+ return (E) elements[(head + logicalIndex) % elements.length];
+ }
+
+ private static long toNanosSaturated(Duration timeout) {
+ try {
+ return timeout.toNanos();
+ } catch (ArithmeticException ignored) {
+ return Long.MAX_VALUE;
+ }
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/CloseableBlockingQueue.java b/src/main/java/dev/azamir/primitives/concurrent/CloseableBlockingQueue.java
new file mode 100644
index 0000000..bff10c8
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/CloseableBlockingQueue.java
@@ -0,0 +1,53 @@
+package dev.azamir.primitives.concurrent;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * A bounded queue contract with explicit shutdown semantics.
+ *
+ * Implementations reject writes after close, allow already accepted elements to drain, and wake or
+ * eventually release blocked callers when the queue lifecycle changes.
+ */
+public interface CloseableBlockingQueue extends AutoCloseable {
+ void put(E element) throws InterruptedException;
+
+ boolean offer(E element, Duration timeout) throws InterruptedException;
+
+ E take() throws InterruptedException;
+
+ Optional poll();
+
+ Optional poll(Duration timeout) throws InterruptedException;
+
+ Optional peek();
+
+ int drainTo(Collection target, int maxElements);
+
+ int clear();
+
+ int closeNow();
+
+ boolean contains(Object value);
+
+ int indexOf(Object value);
+
+ List snapshot();
+
+ boolean isClosed();
+
+ int size();
+
+ default boolean isEmpty() {
+ return size() == 0;
+ }
+
+ int capacity();
+
+ int remainingCapacity();
+
+ @Override
+ void close();
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/CloseableQueues.java b/src/main/java/dev/azamir/primitives/concurrent/CloseableQueues.java
new file mode 100644
index 0000000..7f2e9ac
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/CloseableQueues.java
@@ -0,0 +1,24 @@
+package dev.azamir.primitives.concurrent;
+
+import java.time.Duration;
+import java.util.Objects;
+
+/** Factory methods for the supported bounded queue wait strategies. */
+public final class CloseableQueues {
+ public static final Duration DEFAULT_POLLING_BACKOFF = Duration.ofMillis(10);
+
+ private CloseableQueues() {}
+
+ public static CloseableBlockingQueue bounded(int capacity, QueueWaitStrategy strategy) {
+ return bounded(capacity, strategy, DEFAULT_POLLING_BACKOFF);
+ }
+
+ public static CloseableBlockingQueue bounded(
+ int capacity, QueueWaitStrategy strategy, Duration pollingBackoff) {
+ Objects.requireNonNull(strategy, "strategy");
+ return switch (strategy) {
+ case CONDITION_SIGNALING -> new BoundedBlockingQueue<>(capacity);
+ case POLLING_BACKOFF -> new PollingBackoffQueue<>(capacity, pollingBackoff);
+ };
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/PollingBackoffQueue.java b/src/main/java/dev/azamir/primitives/concurrent/PollingBackoffQueue.java
new file mode 100644
index 0000000..d2a0c6d
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/PollingBackoffQueue.java
@@ -0,0 +1,326 @@
+package dev.azamir.primitives.concurrent;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ * A bounded FIFO queue that uses configurable polling backoff instead of condition signaling.
+ *
+ * This strategy is intentionally less efficient than {@link BoundedBlockingQueue}, but it is useful
+ * when callers need a simple polling lifecycle or want to compare wait strategies. Closing the queue
+ * becomes visible to blocked callers no later than one backoff interval later.
+ */
+public final class PollingBackoffQueue implements CloseableBlockingQueue {
+ private final Object[] elements;
+ private final ReentrantLock lock = new ReentrantLock();
+ private final long backoffNanos;
+
+ private int head;
+ private int tail;
+ private int size;
+ private boolean closed;
+
+ public PollingBackoffQueue(int capacity, Duration backoff) {
+ if (capacity <= 0) {
+ throw new IllegalArgumentException("capacity must be greater than zero");
+ }
+ Objects.requireNonNull(backoff, "backoff");
+ if (backoff.isZero() || backoff.isNegative()) {
+ throw new IllegalArgumentException("backoff must be greater than zero");
+ }
+ elements = new Object[capacity];
+ backoffNanos = toNanosSaturated(backoff);
+ }
+
+ @Override
+ public void put(E element) throws InterruptedException {
+ Objects.requireNonNull(element, "element");
+ while (true) {
+ lock.lockInterruptibly();
+ try {
+ ensureOpenForWrite();
+ if (size < elements.length) {
+ enqueue(element);
+ return;
+ }
+ } finally {
+ lock.unlock();
+ }
+ pause(backoffNanos);
+ }
+ }
+
+ @Override
+ public boolean offer(E element, Duration timeout) throws InterruptedException {
+ Objects.requireNonNull(element, "element");
+ long remaining = validateTimeout(timeout);
+ while (true) {
+ long started = System.nanoTime();
+ lock.lockInterruptibly();
+ try {
+ ensureOpenForWrite();
+ if (size < elements.length) {
+ enqueue(element);
+ return true;
+ }
+ } finally {
+ lock.unlock();
+ }
+ if (remaining <= 0) {
+ return false;
+ }
+ pause(Math.min(backoffNanos, remaining));
+ remaining -= elapsedSince(started);
+ }
+ }
+
+ @Override
+ public E take() throws InterruptedException {
+ while (true) {
+ lock.lockInterruptibly();
+ try {
+ if (size > 0) {
+ return dequeue();
+ }
+ if (closed) {
+ throw new QueueClosedException("queue is closed and drained");
+ }
+ } finally {
+ lock.unlock();
+ }
+ pause(backoffNanos);
+ }
+ }
+
+ @Override
+ public Optional poll() {
+ lock.lock();
+ try {
+ return size == 0 ? Optional.empty() : Optional.of(dequeue());
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public Optional poll(Duration timeout) throws InterruptedException {
+ long remaining = validateTimeout(timeout);
+ while (true) {
+ long started = System.nanoTime();
+ lock.lockInterruptibly();
+ try {
+ if (size > 0) {
+ return Optional.of(dequeue());
+ }
+ if (closed) {
+ return Optional.empty();
+ }
+ } finally {
+ lock.unlock();
+ }
+ if (remaining <= 0) {
+ return Optional.empty();
+ }
+ pause(Math.min(backoffNanos, remaining));
+ remaining -= elapsedSince(started);
+ }
+ }
+
+ @Override
+ public Optional peek() {
+ lock.lock();
+ try {
+ return size == 0 ? Optional.empty() : Optional.of(elementAt(0));
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int drainTo(Collection target, int maxElements) {
+ Objects.requireNonNull(target, "target");
+ if (maxElements < 0) {
+ throw new IllegalArgumentException("maxElements must not be negative");
+ }
+ lock.lock();
+ try {
+ int drained = Math.min(size, maxElements);
+ for (int i = 0; i < drained; i++) {
+ target.add(dequeue());
+ }
+ return drained;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int clear() {
+ lock.lock();
+ try {
+ return clearElements();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int closeNow() {
+ lock.lock();
+ try {
+ closed = true;
+ return clearElements();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public boolean contains(Object value) {
+ return indexOf(value) >= 0;
+ }
+
+ @Override
+ public int indexOf(Object value) {
+ lock.lock();
+ try {
+ for (int i = 0; i < size; i++) {
+ if (Objects.equals(value, elements[(head + i) % elements.length])) {
+ return i;
+ }
+ }
+ return -1;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public List snapshot() {
+ lock.lock();
+ try {
+ List copy = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ copy.add(elementAt(i));
+ }
+ return List.copyOf(copy);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public void close() {
+ lock.lock();
+ try {
+ closed = true;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public boolean isClosed() {
+ lock.lock();
+ try {
+ return closed;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int size() {
+ lock.lock();
+ try {
+ return size;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
+ public int capacity() {
+ return elements.length;
+ }
+
+ @Override
+ public int remainingCapacity() {
+ lock.lock();
+ try {
+ return elements.length - size;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ private void ensureOpenForWrite() {
+ if (closed) {
+ throw new QueueClosedException("queue is closed for writes");
+ }
+ }
+
+ private void enqueue(E element) {
+ elements[tail] = element;
+ tail = (tail + 1) % elements.length;
+ size++;
+ }
+
+ @SuppressWarnings("unchecked")
+ private E dequeue() {
+ E element = (E) elements[head];
+ elements[head] = null;
+ head = (head + 1) % elements.length;
+ size--;
+ return element;
+ }
+
+ private int clearElements() {
+ int removed = size;
+ for (int i = 0; i < removed; i++) {
+ elements[(head + i) % elements.length] = null;
+ }
+ head = 0;
+ tail = 0;
+ size = 0;
+ return removed;
+ }
+
+ @SuppressWarnings("unchecked")
+ private E elementAt(int logicalIndex) {
+ return (E) elements[(head + logicalIndex) % elements.length];
+ }
+
+ private static long validateTimeout(Duration timeout) {
+ Objects.requireNonNull(timeout, "timeout");
+ if (timeout.isNegative()) {
+ throw new IllegalArgumentException("timeout must not be negative");
+ }
+ return toNanosSaturated(timeout);
+ }
+
+ private static void pause(long nanos) throws InterruptedException {
+ if (nanos > 0) {
+ TimeUnit.NANOSECONDS.sleep(nanos);
+ }
+ }
+
+ private static long elapsedSince(long started) {
+ long elapsed = System.nanoTime() - started;
+ return Math.max(elapsed, 1);
+ }
+
+ private static long toNanosSaturated(Duration duration) {
+ try {
+ return duration.toNanos();
+ } catch (ArithmeticException ignored) {
+ return Long.MAX_VALUE;
+ }
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/QueueClosedException.java b/src/main/java/dev/azamir/primitives/concurrent/QueueClosedException.java
new file mode 100644
index 0000000..ec36c78
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/QueueClosedException.java
@@ -0,0 +1,8 @@
+package dev.azamir.primitives.concurrent;
+
+/** Signals that an operation cannot continue because a queue has been closed. */
+public final class QueueClosedException extends IllegalStateException {
+ public QueueClosedException(String message) {
+ super(message);
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/QueueWaitStrategy.java b/src/main/java/dev/azamir/primitives/concurrent/QueueWaitStrategy.java
new file mode 100644
index 0000000..9a642e5
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/QueueWaitStrategy.java
@@ -0,0 +1,7 @@
+package dev.azamir.primitives.concurrent;
+
+/** Selects how a bounded queue waits for capacity or data. */
+public enum QueueWaitStrategy {
+ CONDITION_SIGNALING,
+ POLLING_BACKOFF
+}
diff --git a/src/main/java/dev/azamir/primitives/concurrent/ResettableLongAccumulator.java b/src/main/java/dev/azamir/primitives/concurrent/ResettableLongAccumulator.java
new file mode 100644
index 0000000..414a8ac
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/concurrent/ResettableLongAccumulator.java
@@ -0,0 +1,47 @@
+package dev.azamir.primitives.concurrent;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.LongBinaryOperator;
+
+/**
+ * A lock-free long accumulator with an atomic reset operation.
+ *
+ * The supplied function should be side-effect free and associative because it may be invoked multiple
+ * times when concurrent updates retry.
+ */
+public final class ResettableLongAccumulator {
+ private final LongBinaryOperator function;
+ private final long identity;
+ private final AtomicLong value;
+
+ public ResettableLongAccumulator(LongBinaryOperator function, long identity) {
+ this.function = Objects.requireNonNull(function, "function");
+ this.identity = identity;
+ value = new AtomicLong(identity);
+ }
+
+ public static ResettableLongAccumulator sum() {
+ return new ResettableLongAccumulator(Long::sum, 0);
+ }
+
+ public static ResettableLongAccumulator max(long identity) {
+ return new ResettableLongAccumulator(Math::max, identity);
+ }
+
+ public long accumulate(long operand) {
+ return value.accumulateAndGet(operand, function);
+ }
+
+ public long get() {
+ return value.get();
+ }
+
+ public long getAndReset() {
+ return value.getAndSet(identity);
+ }
+
+ public void reset() {
+ value.set(identity);
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/math/DecimalRatioCalculator.java b/src/main/java/dev/azamir/primitives/math/DecimalRatioCalculator.java
new file mode 100644
index 0000000..b1a40db
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/math/DecimalRatioCalculator.java
@@ -0,0 +1,60 @@
+package dev.azamir.primitives.math;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.Objects;
+
+/**
+ * Calculates decimal ratios with explicit scale, rounding, and an optional additive adjustment.
+ *
+ *
The adjustment is applied after division, matching use cases where a calibrated factor must be
+ * added to a measured ratio.
+ */
+public final class DecimalRatioCalculator {
+ private final BigDecimal adjustment;
+
+ public DecimalRatioCalculator() {
+ this(BigDecimal.ZERO);
+ }
+
+ public DecimalRatioCalculator(BigDecimal adjustment) {
+ this.adjustment = Objects.requireNonNull(adjustment, "adjustment");
+ }
+
+ public BigDecimal calculate(BigDecimal numerator, BigDecimal denominator) {
+ return calculate(numerator, denominator, 1, RoundingMode.UP);
+ }
+
+ public BigDecimal calculate(BigDecimal numerator, BigDecimal denominator, int scale) {
+ return calculate(numerator, denominator, scale, RoundingMode.UP);
+ }
+
+ public BigDecimal calculate(
+ BigDecimal numerator, BigDecimal denominator, RoundingMode roundingMode) {
+ return calculate(numerator, denominator, 1, roundingMode);
+ }
+
+ public BigDecimal calculate(
+ BigDecimal numerator, BigDecimal denominator, int scale, RoundingMode roundingMode) {
+ Objects.requireNonNull(numerator, "numerator");
+ Objects.requireNonNull(denominator, "denominator");
+ Objects.requireNonNull(roundingMode, "roundingMode");
+ if (scale < 0) {
+ throw new IllegalArgumentException("scale must not be negative");
+ }
+ if (denominator.signum() == 0) {
+ throw new ArithmeticException("denominator must not be zero");
+ }
+ return numerator.divide(denominator, scale, roundingMode).add(adjustment);
+ }
+
+ public double calculate(double numerator, double denominator, int scale, RoundingMode roundingMode) {
+ return calculate(
+ BigDecimal.valueOf(numerator), BigDecimal.valueOf(denominator), scale, roundingMode)
+ .doubleValue();
+ }
+
+ public BigDecimal adjustment() {
+ return adjustment;
+ }
+}
diff --git a/src/main/java/dev/azamir/primitives/statistics/StatisticsSnapshot.java b/src/main/java/dev/azamir/primitives/statistics/StatisticsSnapshot.java
new file mode 100644
index 0000000..d1d2ff9
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/statistics/StatisticsSnapshot.java
@@ -0,0 +1,13 @@
+package dev.azamir.primitives.statistics;
+
+import java.util.OptionalDouble;
+
+/** Immutable view of a streaming statistics accumulator. */
+public record StatisticsSnapshot(
+ long count,
+ double sum,
+ OptionalDouble mean,
+ OptionalDouble sampleVariance,
+ OptionalDouble sampleStandardDeviation,
+ OptionalDouble min,
+ OptionalDouble max) {}
diff --git a/src/main/java/dev/azamir/primitives/statistics/StreamingStatistics.java b/src/main/java/dev/azamir/primitives/statistics/StreamingStatistics.java
new file mode 100644
index 0000000..d9cc82f
--- /dev/null
+++ b/src/main/java/dev/azamir/primitives/statistics/StreamingStatistics.java
@@ -0,0 +1,98 @@
+package dev.azamir.primitives.statistics;
+
+import java.util.OptionalDouble;
+
+/**
+ * Thread-safe one-pass descriptive statistics using Welford's algorithm for variance and compensated
+ * summation for improved numerical stability.
+ */
+public final class StreamingStatistics {
+ private long count;
+ private double mean;
+ private double squaredDeviationSum;
+ private double sum;
+ private double sumCompensation;
+ private double min = Double.POSITIVE_INFINITY;
+ private double max = Double.NEGATIVE_INFINITY;
+
+ public synchronized void add(double value) {
+ if (!Double.isFinite(value)) {
+ throw new IllegalArgumentException("value must be finite");
+ }
+
+ count++;
+ double delta = value - mean;
+ mean += delta / count;
+ double adjustedDelta = value - mean;
+ squaredDeviationSum += delta * adjustedDelta;
+
+ double compensatedValue = value - sumCompensation;
+ double updatedSum = sum + compensatedValue;
+ sumCompensation = (updatedSum - sum) - compensatedValue;
+ sum = updatedSum;
+
+ min = Math.min(min, value);
+ max = Math.max(max, value);
+ }
+
+ public synchronized StatisticsSnapshot snapshot() {
+ OptionalDouble currentMean = count == 0 ? OptionalDouble.empty() : OptionalDouble.of(mean);
+ OptionalDouble variance =
+ count < 2 ? OptionalDouble.empty() : OptionalDouble.of(squaredDeviationSum / (count - 1));
+ OptionalDouble standardDeviation =
+ variance.isEmpty() ? OptionalDouble.empty() : OptionalDouble.of(Math.sqrt(variance.getAsDouble()));
+ OptionalDouble currentMin = count == 0 ? OptionalDouble.empty() : OptionalDouble.of(min);
+ OptionalDouble currentMax = count == 0 ? OptionalDouble.empty() : OptionalDouble.of(max);
+
+ return new StatisticsSnapshot(
+ count, sum, currentMean, variance, standardDeviation, currentMin, currentMax);
+ }
+
+ public synchronized long count() {
+ return count;
+ }
+
+ public synchronized double sum() {
+ return sum;
+ }
+
+ public synchronized OptionalDouble mean() {
+ return count == 0 ? OptionalDouble.empty() : OptionalDouble.of(mean);
+ }
+
+ public synchronized OptionalDouble sampleVariance() {
+ return count < 2
+ ? OptionalDouble.empty()
+ : OptionalDouble.of(squaredDeviationSum / (count - 1));
+ }
+
+ public synchronized OptionalDouble sampleStandardDeviation() {
+ OptionalDouble variance = sampleVariance();
+ return variance.isEmpty()
+ ? OptionalDouble.empty()
+ : OptionalDouble.of(Math.sqrt(variance.getAsDouble()));
+ }
+
+ public synchronized OptionalDouble min() {
+ return count == 0 ? OptionalDouble.empty() : OptionalDouble.of(min);
+ }
+
+ public synchronized OptionalDouble max() {
+ return count == 0 ? OptionalDouble.empty() : OptionalDouble.of(max);
+ }
+
+ public synchronized void reset() {
+ count = 0;
+ mean = 0;
+ squaredDeviationSum = 0;
+ sum = 0;
+ sumCompensation = 0;
+ min = Double.POSITIVE_INFINITY;
+ max = Double.NEGATIVE_INFINITY;
+ }
+
+ @Override
+ public synchronized String toString() {
+ return snapshot().toString();
+ }
+}
diff --git a/src/test/java/dev/azamir/primitives/batching/AdaptiveBatchSizerTest.java b/src/test/java/dev/azamir/primitives/batching/AdaptiveBatchSizerTest.java
new file mode 100644
index 0000000..5f9afa9
--- /dev/null
+++ b/src/test/java/dev/azamir/primitives/batching/AdaptiveBatchSizerTest.java
@@ -0,0 +1,64 @@
+package dev.azamir.primitives.batching;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import org.junit.jupiter.api.Test;
+
+class AdaptiveBatchSizerTest {
+ @Test
+ void estimatesRecordsFromPayloadAndFixedOverhead() {
+ AdaptiveBatchSizer sizer = new AdaptiveBatchSizer(1_000, 0.8);
+
+ assertEquals(18, sizer.estimate(10, 500, 100));
+ }
+
+ @Test
+ void fallsBackToSampleSizeWhenOverheadOrPayloadPreventsProjection() {
+ AdaptiveBatchSizer sizer = new AdaptiveBatchSizer(1_000, 0.8);
+
+ assertEquals(10, sizer.estimate(10, 1_200, 1_100));
+ assertEquals(10, sizer.estimate(10, 100, 100));
+ assertEquals(0, sizer.estimate(0, 0, 0));
+ }
+
+ @Test
+ void clampsProjectedBatchSize() {
+ AdaptiveBatchSizer sizer = new AdaptiveBatchSizer(10_000, 1.0, 5, 50);
+
+ assertEquals(50, sizer.estimate(10, 20, 10));
+ }
+
+ @Test
+ void retriesTransientProbeFailuresBeforeEstimating() {
+ AtomicInteger attempts = new AtomicInteger();
+ RetryingSizeProbe probe =
+ new RetryingSizeProbe<>(
+ ignored -> {
+ if (attempts.incrementAndGet() < 3) {
+ throw new IllegalStateException("temporary serialization failure");
+ }
+ return 500;
+ },
+ 3);
+ AdaptiveBatchSizer sizer = new AdaptiveBatchSizer(1_000, 0.8);
+
+ assertEquals(18, sizer.estimate(10, "sample", probe, 100));
+ assertEquals(3, attempts.get());
+ }
+
+ @Test
+ void reportsExhaustedRetryBudget() {
+ RetryingSizeProbe probe =
+ new RetryingSizeProbe<>(ignored -> { throw new IllegalStateException("failed"); }, 2);
+
+ assertThrows(SizeProbeException.class, () -> probe.measureBytes("sample"));
+ }
+
+ @Test
+ void rejectsInconsistentMeasurements() {
+ AdaptiveBatchSizer sizer = new AdaptiveBatchSizer(1_000, 0.8);
+ assertThrows(IllegalArgumentException.class, () -> sizer.estimate(10, 99, 100));
+ }
+}
diff --git a/src/test/java/dev/azamir/primitives/concurrent/BoundedBlockingQueueTest.java b/src/test/java/dev/azamir/primitives/concurrent/BoundedBlockingQueueTest.java
new file mode 100644
index 0000000..91e8ec5
--- /dev/null
+++ b/src/test/java/dev/azamir/primitives/concurrent/BoundedBlockingQueueTest.java
@@ -0,0 +1,212 @@
+package dev.azamir.primitives.concurrent;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.junit.jupiter.api.Test;
+
+class BoundedBlockingQueueTest {
+ @Test
+ void preservesFifoOrderAndCapacityAccounting() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(3);
+
+ queue.put(10);
+ queue.put(20);
+ queue.put(30);
+
+ assertEquals(3, queue.size());
+ assertEquals(0, queue.remainingCapacity());
+ assertEquals(10, queue.take());
+ assertEquals(20, queue.take());
+ assertEquals(30, queue.take());
+ assertEquals(0, queue.size());
+ assertEquals(3, queue.remainingCapacity());
+ }
+
+ @Test
+ void producerWaitsUntilCapacityIsAvailable() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(1);
+ queue.put(1);
+
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CountDownLatch started = new CountDownLatch(1);
+ try {
+ Future producer =
+ executor.submit(
+ () -> {
+ started.countDown();
+ queue.put(2);
+ return null;
+ });
+
+ assertTrue(started.await(1, SECONDS));
+ Thread.sleep(100);
+ assertFalse(producer.isDone());
+
+ assertEquals(1, queue.take());
+ producer.get(2, SECONDS);
+ assertEquals(2, queue.take());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void consumerWaitsUntilAnElementArrives() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ CountDownLatch started = new CountDownLatch(1);
+ try {
+ Future consumer =
+ executor.submit(
+ () -> {
+ started.countDown();
+ return queue.take();
+ });
+
+ assertTrue(started.await(1, SECONDS));
+ Thread.sleep(100);
+ assertFalse(consumer.isDone());
+
+ queue.put(42);
+ assertEquals(42, consumer.get(2, SECONDS));
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void closeRejectsWritesButAllowsAcceptedElementsToDrain() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(2);
+ queue.put(1);
+ queue.put(2);
+
+ queue.close();
+
+ assertTrue(queue.isClosed());
+ assertThrows(QueueClosedException.class, () -> queue.put(3));
+ assertEquals(1, queue.take());
+ assertEquals(2, queue.take());
+ assertThrows(QueueClosedException.class, queue::take);
+ assertTrue(queue.poll().isEmpty());
+ }
+
+ @Test
+ void closeWakesBlockedConsumers() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future consumer = executor.submit(queue::take);
+ Thread.sleep(100);
+ queue.close();
+
+ ExecutionException error =
+ assertThrows(ExecutionException.class, () -> consumer.get(2, SECONDS));
+ assertInstanceOf(QueueClosedException.class, error.getCause());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void timedOperationsRespectTimeouts() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(1);
+ queue.put(1);
+
+ assertFalse(queue.offer(2, Duration.ofMillis(30)));
+ assertEquals(1, queue.take());
+ assertTrue(queue.poll(Duration.ofMillis(30)).isEmpty());
+ }
+
+ @Test
+ void drainToRemovesAtMostRequestedElements() throws Exception {
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(4);
+ queue.put(1);
+ queue.put(2);
+ queue.put(3);
+ List target = new ArrayList<>();
+
+ assertEquals(2, queue.drainTo(target, 2));
+ assertEquals(List.of(1, 2), target);
+ assertEquals(1, queue.size());
+ assertEquals(3, queue.take());
+ }
+
+ @Test
+ void concurrentProducersAndConsumersDoNotLoseOrDuplicateElements() throws Exception {
+ int producers = 4;
+ int consumers = 4;
+ int elementsPerProducer = 2_000;
+ int expected = producers * elementsPerProducer;
+ BoundedBlockingQueue queue = new BoundedBlockingQueue<>(128);
+ ExecutorService executor = Executors.newFixedThreadPool(producers + consumers);
+ Set consumed = ConcurrentHashMap.newKeySet();
+ AtomicBoolean duplicate = new AtomicBoolean();
+ List> consumerFutures = new ArrayList<>();
+ List> producerFutures = new ArrayList<>();
+
+ try {
+ for (int i = 0; i < consumers; i++) {
+ consumerFutures.add(
+ executor.submit(
+ () -> {
+ try {
+ while (true) {
+ if (!consumed.add(queue.take())) {
+ duplicate.set(true);
+ }
+ }
+ } catch (QueueClosedException ignored) {
+ // Expected after all accepted elements are drained.
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ }
+ }));
+ }
+
+ for (int producer = 0; producer < producers; producer++) {
+ int producerId = producer;
+ producerFutures.add(
+ executor.submit(
+ () -> {
+ for (int i = 0; i < elementsPerProducer; i++) {
+ queue.put(producerId * elementsPerProducer + i);
+ }
+ return null;
+ }));
+ }
+
+ for (Future future : producerFutures) {
+ future.get(10, SECONDS);
+ }
+ queue.close();
+ for (Future future : consumerFutures) {
+ future.get(10, SECONDS);
+ }
+
+ assertFalse(duplicate.get());
+ assertEquals(expected, consumed.size());
+ for (int value = 0; value < expected; value++) {
+ assertTrue(consumed.contains(value));
+ }
+ } finally {
+ queue.close();
+ executor.shutdownNow();
+ }
+ }
+}
diff --git a/src/test/java/dev/azamir/primitives/concurrent/QueueStrategiesTest.java b/src/test/java/dev/azamir/primitives/concurrent/QueueStrategiesTest.java
new file mode 100644
index 0000000..4c697f6
--- /dev/null
+++ b/src/test/java/dev/azamir/primitives/concurrent/QueueStrategiesTest.java
@@ -0,0 +1,104 @@
+package dev.azamir.primitives.concurrent;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import org.junit.jupiter.api.Test;
+
+class QueueStrategiesTest {
+ @Test
+ void factoryCreatesBothSupportedWaitStrategies() {
+ assertInstanceOf(
+ BoundedBlockingQueue.class,
+ CloseableQueues.bounded(4, QueueWaitStrategy.CONDITION_SIGNALING));
+ assertInstanceOf(
+ PollingBackoffQueue.class,
+ CloseableQueues.bounded(
+ 4, QueueWaitStrategy.POLLING_BACKOFF, Duration.ofMillis(5)));
+ }
+
+ @Test
+ void queueCollectionOperationsUseLogicalFifoOrder() throws Exception {
+ CloseableBlockingQueue queue = new BoundedBlockingQueue<>(4);
+ queue.put(10);
+ queue.put(20);
+ queue.put(30);
+
+ assertFalse(queue.isEmpty());
+ assertEquals(10, queue.peek().orElseThrow());
+ assertEquals(List.of(10, 20, 30), queue.snapshot());
+ assertTrue(queue.contains(20));
+ assertEquals(1, queue.indexOf(20));
+ assertEquals(-1, queue.indexOf(99));
+ assertEquals(3, queue.clear());
+ assertTrue(queue.isEmpty());
+ assertEquals(List.of(), queue.snapshot());
+ assertEquals(4, queue.remainingCapacity());
+ }
+
+ @Test
+ void closeNowAtomicallyDiscardsBufferedDataAndRejectsWrites() throws Exception {
+ CloseableBlockingQueue queue = new BoundedBlockingQueue<>(3);
+ queue.put(1);
+ queue.put(2);
+
+ assertEquals(2, queue.closeNow());
+ assertTrue(queue.isClosed());
+ assertTrue(queue.isEmpty());
+ assertTrue(queue.peek().isEmpty());
+ assertThrows(QueueClosedException.class, () -> queue.put(3));
+ assertThrows(QueueClosedException.class, queue::take);
+ }
+
+ @Test
+ void pollingProducerWaitsUntilCapacityBecomesAvailable() throws Exception {
+ CloseableBlockingQueue queue =
+ new PollingBackoffQueue<>(1, Duration.ofMillis(5));
+ queue.put(1);
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future producer =
+ executor.submit(
+ () -> {
+ queue.put(2);
+ return null;
+ });
+ Thread.sleep(30);
+ assertEquals(1, queue.take());
+ producer.get(2, SECONDS);
+ assertEquals(2, queue.take());
+ } finally {
+ queue.close();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void pollingCloseReleasesBlockedConsumerWithinBackoffInterval() throws Exception {
+ CloseableBlockingQueue queue =
+ new PollingBackoffQueue<>(1, Duration.ofMillis(5));
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future consumer = executor.submit(queue::take);
+ Thread.sleep(30);
+ queue.close();
+
+ ExecutionException failure =
+ assertThrows(ExecutionException.class, () -> consumer.get(2, SECONDS));
+ assertInstanceOf(QueueClosedException.class, failure.getCause());
+ } finally {
+ queue.close();
+ executor.shutdownNow();
+ }
+ }
+}
diff --git a/src/test/java/dev/azamir/primitives/concurrent/ResettableLongAccumulatorTest.java b/src/test/java/dev/azamir/primitives/concurrent/ResettableLongAccumulatorTest.java
new file mode 100644
index 0000000..b4182c4
--- /dev/null
+++ b/src/test/java/dev/azamir/primitives/concurrent/ResettableLongAccumulatorTest.java
@@ -0,0 +1,64 @@
+package dev.azamir.primitives.concurrent;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import org.junit.jupiter.api.Test;
+
+class ResettableLongAccumulatorTest {
+ @Test
+ void accumulatesAndResetsAtomically() {
+ ResettableLongAccumulator accumulator = ResettableLongAccumulator.sum();
+
+ accumulator.accumulate(10);
+ accumulator.accumulate(15);
+
+ assertEquals(25, accumulator.get());
+ assertEquals(25, accumulator.getAndReset());
+ assertEquals(0, accumulator.get());
+ }
+
+ @Test
+ void supportsCustomAssociativeFunctions() {
+ ResettableLongAccumulator maximum = ResettableLongAccumulator.max(Long.MIN_VALUE);
+
+ maximum.accumulate(10);
+ maximum.accumulate(4);
+ maximum.accumulate(30);
+
+ assertEquals(30, maximum.get());
+ }
+
+ @Test
+ void concurrentUpdatesAreNotLost() throws Exception {
+ int workers = 8;
+ int incrementsPerWorker = 50_000;
+ ResettableLongAccumulator accumulator = ResettableLongAccumulator.sum();
+ ExecutorService executor = Executors.newFixedThreadPool(workers);
+ List> futures = new ArrayList<>();
+
+ try {
+ for (int worker = 0; worker < workers; worker++) {
+ futures.add(
+ executor.submit(
+ () -> {
+ for (int i = 0; i < incrementsPerWorker; i++) {
+ accumulator.accumulate(1);
+ }
+ }));
+ }
+ for (Future future : futures) {
+ future.get(10, SECONDS);
+ }
+
+ assertEquals((long) workers * incrementsPerWorker, accumulator.get());
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+}
diff --git a/src/test/java/dev/azamir/primitives/math/DecimalRatioCalculatorTest.java b/src/test/java/dev/azamir/primitives/math/DecimalRatioCalculatorTest.java
new file mode 100644
index 0000000..5eb3a26
--- /dev/null
+++ b/src/test/java/dev/azamir/primitives/math/DecimalRatioCalculatorTest.java
@@ -0,0 +1,45 @@
+package dev.azamir.primitives.math;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import org.junit.jupiter.api.Test;
+
+class DecimalRatioCalculatorTest {
+ @Test
+ void appliesScaleRoundingAndAdjustment() {
+ DecimalRatioCalculator calculator = new DecimalRatioCalculator(new BigDecimal("0.10"));
+
+ assertEquals(
+ new BigDecimal("0.77"),
+ calculator.calculate(
+ new BigDecimal("2"), new BigDecimal("3"), 2, RoundingMode.HALF_UP));
+ }
+
+ @Test
+ void preservesDefaultUpRoundingBehavior() {
+ DecimalRatioCalculator calculator = new DecimalRatioCalculator();
+ assertEquals(
+ new BigDecimal("0.7"),
+ calculator.calculate(new BigDecimal("2"), new BigDecimal("3")));
+ }
+
+ @Test
+ void supportsDoubleConvenienceMethodWithoutBinaryConstructorArtifacts() {
+ DecimalRatioCalculator calculator = new DecimalRatioCalculator(new BigDecimal("0.05"));
+ assertEquals(0.72, calculator.calculate(2, 3, 2, RoundingMode.HALF_UP), 1e-12);
+ }
+
+ @Test
+ void rejectsInvalidScaleAndZeroDenominator() {
+ DecimalRatioCalculator calculator = new DecimalRatioCalculator();
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> calculator.calculate(BigDecimal.ONE, BigDecimal.TEN, -1, RoundingMode.UP));
+ assertThrows(
+ ArithmeticException.class,
+ () -> calculator.calculate(BigDecimal.ONE, BigDecimal.ZERO, 2, RoundingMode.UP));
+ }
+}
diff --git a/src/test/java/dev/azamir/primitives/statistics/StreamingStatisticsTest.java b/src/test/java/dev/azamir/primitives/statistics/StreamingStatisticsTest.java
new file mode 100644
index 0000000..0bc919e
--- /dev/null
+++ b/src/test/java/dev/azamir/primitives/statistics/StreamingStatisticsTest.java
@@ -0,0 +1,84 @@
+package dev.azamir.primitives.statistics;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import org.junit.jupiter.api.Test;
+
+class StreamingStatisticsTest {
+ @Test
+ void calculatesStableOnePassStatistics() {
+ StreamingStatistics statistics = new StreamingStatistics();
+ for (double value : new double[] {2, 4, 4, 4, 5, 5, 7, 9}) {
+ statistics.add(value);
+ }
+
+ StatisticsSnapshot snapshot = statistics.snapshot();
+ assertEquals(8, snapshot.count());
+ assertEquals(40, snapshot.sum(), 1e-12);
+ assertEquals(5, snapshot.mean().orElseThrow(), 1e-12);
+ assertEquals(32.0 / 7.0, snapshot.sampleVariance().orElseThrow(), 1e-12);
+ assertEquals(Math.sqrt(32.0 / 7.0), snapshot.sampleStandardDeviation().orElseThrow(), 1e-12);
+ assertEquals(2, snapshot.min().orElseThrow(), 1e-12);
+ assertEquals(9, snapshot.max().orElseThrow(), 1e-12);
+ }
+
+ @Test
+ void exposesEmptyStateAndSupportsReset() {
+ StreamingStatistics statistics = new StreamingStatistics();
+ assertTrue(statistics.mean().isEmpty());
+ assertTrue(statistics.sampleVariance().isEmpty());
+ assertTrue(statistics.min().isEmpty());
+ assertTrue(statistics.max().isEmpty());
+
+ statistics.add(12.5);
+ statistics.reset();
+
+ assertEquals(0, statistics.count());
+ assertEquals(0, statistics.sum(), 0);
+ assertTrue(statistics.snapshot().mean().isEmpty());
+ }
+
+ @Test
+ void rejectsNonFiniteValues() {
+ StreamingStatistics statistics = new StreamingStatistics();
+ assertThrows(IllegalArgumentException.class, () -> statistics.add(Double.NaN));
+ assertThrows(IllegalArgumentException.class, () -> statistics.add(Double.POSITIVE_INFINITY));
+ }
+
+ @Test
+ void acceptsConcurrentUpdatesWithoutLosingValues() throws Exception {
+ StreamingStatistics statistics = new StreamingStatistics();
+ ExecutorService executor = Executors.newFixedThreadPool(4);
+ try {
+ List> futures =
+ List.of(
+ executor.submit(() -> addRepeatedly(statistics, 1_000)),
+ executor.submit(() -> addRepeatedly(statistics, 1_000)),
+ executor.submit(() -> addRepeatedly(statistics, 1_000)),
+ executor.submit(() -> addRepeatedly(statistics, 1_000)));
+ for (Future future : futures) {
+ future.get(5, SECONDS);
+ }
+
+ assertEquals(4_000, statistics.count());
+ assertEquals(4_000, statistics.sum(), 1e-12);
+ assertEquals(1, statistics.mean().orElseThrow(), 1e-12);
+ assertEquals(0, statistics.sampleVariance().orElseThrow(), 1e-12);
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private static void addRepeatedly(StreamingStatistics statistics, int count) {
+ for (int i = 0; i < count; i++) {
+ statistics.add(1);
+ }
+ }
+}