Each message in a batch is independently validated and processed. A failed message does not + * affect the others. Use {@link #isSuccess()} to check the outcome, then either {@link + * #getMessageId()} or {@link #getError()}. + */ +public final class BatchEnqueueResult { + private final String messageId; + private final String error; + + private BatchEnqueueResult(String messageId, String error) { + this.messageId = messageId; + this.error = error; + } + + /** Create a successful result with the broker-assigned message ID. */ + static BatchEnqueueResult success(String messageId) { + return new BatchEnqueueResult(messageId, null); + } + + /** Create a failed result with an error description. */ + static BatchEnqueueResult error(String error) { + return new BatchEnqueueResult(null, error); + } + + /** Returns true if the message was successfully enqueued. */ + public boolean isSuccess() { + return messageId != null; + } + + /** + * Returns the broker-assigned message ID. + * + * @throws IllegalStateException if this result is an error + */ + public String getMessageId() { + if (messageId == null) { + throw new IllegalStateException("result is an error: " + error); + } + return messageId; + } + + /** + * Returns the error description. + * + * @throws IllegalStateException if this result is a success + */ + public String getError() { + if (error == null) { + throw new IllegalStateException("result is a success"); + } + return error; + } +} diff --git a/src/main/java/dev/faisca/fila/BatchMode.java b/src/main/java/dev/faisca/fila/BatchMode.java new file mode 100644 index 0000000..0be14fc --- /dev/null +++ b/src/main/java/dev/faisca/fila/BatchMode.java @@ -0,0 +1,93 @@ +package dev.faisca.fila; + +/** + * Controls how the SDK batches {@link FilaClient#enqueue} calls. + * + *
The default is {@link #auto()} -- opportunistic batching that requires zero configuration. At + * low load each message is sent individually (zero added latency). At high load messages accumulate + * naturally and are flushed together. + */ +public final class BatchMode { + enum Kind { + AUTO, + LINGER, + DISABLED + } + + private final Kind kind; + private final int maxBatchSize; + private final long lingerMs; + + private BatchMode(Kind kind, int maxBatchSize, long lingerMs) { + this.kind = kind; + this.maxBatchSize = maxBatchSize; + this.lingerMs = lingerMs; + } + + /** + * Opportunistic batching (default). + * + *
A background thread blocks for the first message, then drains any additional messages that + * arrived concurrently. At low load each message is sent individually. At high load messages + * accumulate naturally into batches. Zero config, zero latency penalty. + * + * @return a new AUTO batch mode with default max batch size (100) + */ + public static BatchMode auto() { + return new BatchMode(Kind.AUTO, 100, 0); + } + + /** + * Opportunistic batching with a custom max batch size. + * + * @param maxBatchSize safety cap on batch size + * @return a new AUTO batch mode + */ + public static BatchMode auto(int maxBatchSize) { + if (maxBatchSize < 1) { + throw new IllegalArgumentException("maxBatchSize must be >= 1"); + } + return new BatchMode(Kind.AUTO, maxBatchSize, 0); + } + + /** + * Timer-based forced batching. + * + *
Buffers messages and flushes when either {@code batchSize} messages accumulate or {@code + * lingerMs} milliseconds elapse since the first message in the batch -- whichever comes first. + * + * @param lingerMs time threshold in milliseconds before a partial batch is flushed + * @param batchSize maximum messages per batch + * @return a new LINGER batch mode + */ + public static BatchMode linger(long lingerMs, int batchSize) { + if (lingerMs < 1) { + throw new IllegalArgumentException("lingerMs must be >= 1"); + } + if (batchSize < 1) { + throw new IllegalArgumentException("batchSize must be >= 1"); + } + return new BatchMode(Kind.LINGER, batchSize, lingerMs); + } + + /** + * No batching. Each {@link FilaClient#enqueue} call is a direct single-message RPC. + * + * @return a DISABLED batch mode + */ + public static BatchMode disabled() { + return new BatchMode(Kind.DISABLED, 0, 0); + } + + Kind getKind() { + return kind; + } + + int getMaxBatchSize() { + return maxBatchSize; + } + + long getLingerMs() { + return lingerMs; + } +} diff --git a/src/main/java/dev/faisca/fila/Batcher.java b/src/main/java/dev/faisca/fila/Batcher.java new file mode 100644 index 0000000..486f745 --- /dev/null +++ b/src/main/java/dev/faisca/fila/Batcher.java @@ -0,0 +1,288 @@ +package dev.faisca.fila; + +import fila.v1.FilaServiceGrpc; +import fila.v1.Service; +import io.grpc.StatusRuntimeException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +/** + * Background batcher that coalesces individual enqueue calls into batch RPCs. + * + *
Supports two modes: AUTO (opportunistic, Nagle-style) and LINGER (timer-based). The batcher
+ * runs on a dedicated daemon thread and flushes RPCs on an executor pool.
+ */
+final class Batcher {
+ private final LinkedBlockingQueue Each message specifies its target queue, headers, and payload independently, allowing a single
+ * batch to target multiple queues.
+ */
+public final class EnqueueMessage {
+ private final String queue;
+ private final Map Wraps the hot-path gRPC operations: enqueue, consume, ack, nack.
+ * Wraps the hot-path gRPC operations: enqueue, batch enqueue, consume, ack, nack.
+ *
+ * By default, {@code enqueue()} routes through an opportunistic batcher that coalesces messages
+ * at high load without adding latency at low load. Use {@link Builder#withBatchMode(BatchMode)} to
+ * configure batching behavior.
*
* When batching is enabled (the default), the message is submitted to the background batcher
+ * and may be coalesced with other messages. The method blocks until the message is acknowledged
+ * by the broker.
+ *
* @param queue target queue name
* @param headers message headers (may be empty)
* @param payload message payload bytes
@@ -76,24 +91,78 @@ public static Builder builder(String address) {
* @throws RpcException for unexpected gRPC failures
*/
public String enqueue(String queue, Map Each message is independently validated and processed. A failed message does not affect the
+ * others in the batch. Returns a list of results with one entry per input message, in the same
+ * order.
+ *
+ * This bypasses the batcher and always uses the {@code BatchEnqueue} RPC directly.
+ *
+ * @param messages the messages to enqueue
+ * @return a list of results, one per input message
+ * @throws RpcException for transport-level failures affecting the entire batch
+ */
+ public List Messages are delivered to the handler on a background thread. Nacked messages are
+ * Messages are delivered to the handler on a background thread. The handler transparently
+ * receives messages from both singular and batched server responses. Nacked messages are
* redelivered on the same stream. Call {@link ConsumerHandle#cancel()} to stop consuming.
*
* @param queue queue to consume from
@@ -174,9 +243,16 @@ public void nack(String queue, String msgId, String error) {
}
}
- /** Shut down the underlying gRPC channel. */
+ /**
+ * Shut down the client, draining any pending batched messages before disconnecting.
+ *
+ * If a batcher is running, pending messages are flushed before the gRPC channel is closed.
+ */
@Override
public void close() {
+ if (batcher != null) {
+ batcher.shutdown();
+ }
channel.shutdown();
try {
if (!channel.awaitTermination(5, TimeUnit.SECONDS)) {
@@ -188,14 +264,46 @@ public void close() {
}
}
+ /** Direct single-message enqueue RPC (no batcher). */
+ private String enqueueDirect(String queue, Map Prefers the batched {@code messages} field when non-empty, falling back to the singular
+ * {@code message} field for backward compatibility with older servers.
+ */
private static void consumeStream(
Iterator Default is {@link BatchMode#auto()} -- opportunistic batching. Use {@link
+ * BatchMode#disabled()} to turn off batching entirely.
+ *
+ * @param batchMode the batch mode
+ * @return this builder
+ */
+ public Builder withBatchMode(BatchMode batchMode) {
+ this.batchMode = batchMode;
+ return this;
+ }
+
/** Build and connect the client. */
public FilaClient build() {
if (clientCertPem != null && !tlsEnabled) {
@@ -447,7 +578,19 @@ public FilaClient build() {
channel = channelBuilder.build();
}
- return new FilaClient(channel, caCertPem, clientCertPem, clientKeyPem, apiKey);
+ Batcher batcherInstance = null;
+ if (batchMode.getKind() != BatchMode.Kind.DISABLED) {
+ FilaServiceGrpc.FilaServiceBlockingStub batcherStub =
+ FilaServiceGrpc.newBlockingStub(channel);
+ if (apiKey != null) {
+ // The stub needs the interceptor applied at channel level (already done above).
+ // No additional interceptor needed on the stub.
+ }
+ batcherInstance = new Batcher(batcherStub, batchMode);
+ }
+
+ return new FilaClient(
+ channel, caCertPem, clientCertPem, clientKeyPem, apiKey, batcherInstance);
}
static String parseHost(String address) {
diff --git a/src/test/java/dev/faisca/fila/BatchClientTest.java b/src/test/java/dev/faisca/fila/BatchClientTest.java
new file mode 100644
index 0000000..6338b58
--- /dev/null
+++ b/src/test/java/dev/faisca/fila/BatchClientTest.java
@@ -0,0 +1,241 @@
+package dev.faisca.fila;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
+
+/**
+ * Integration tests for batch enqueue and smart batching.
+ *
+ * Requires a fila-server binary. Skipped if not available.
+ */
+@EnabledIf("serverAvailable")
+class BatchClientTest {
+ private static TestServer server;
+
+ @BeforeAll
+ static void setUp() throws Exception {
+ server = TestServer.start();
+ server.createQueue("test-batch-explicit");
+ server.createQueue("test-batch-auto");
+ server.createQueue("test-batch-linger");
+ server.createQueue("test-batch-disabled");
+ server.createQueue("test-batch-consume");
+ server.createQueue("test-batch-mixed");
+ }
+
+ @AfterAll
+ static void tearDown() {
+ if (server != null) server.stop();
+ }
+
+ static boolean serverAvailable() {
+ return TestServer.isBinaryAvailable();
+ }
+
+ @Test
+ void explicitBatchEnqueue() {
+ try (FilaClient client =
+ FilaClient.builder(server.address()).withBatchMode(BatchMode.disabled()).build()) {
+ List{@code
* try (FilaClient client = FilaClient.builder("localhost:5555").build()) {
@@ -45,19 +53,22 @@ public final class FilaClient implements AutoCloseable {
private final byte[] clientCertPem;
private final byte[] clientKeyPem;
private final String apiKey;
+ private final Batcher batcher;
private FilaClient(
ManagedChannel channel,
byte[] caCertPem,
byte[] clientCertPem,
byte[] clientKeyPem,
- String apiKey) {
+ String apiKey,
+ Batcher batcher) {
this.channel = channel;
this.blockingStub = FilaServiceGrpc.newBlockingStub(channel);
this.caCertPem = caCertPem;
this.clientCertPem = clientCertPem;
this.clientKeyPem = clientKeyPem;
this.apiKey = apiKey;
+ this.batcher = batcher;
}
/** Returns a new builder for configuring a {@link FilaClient}. */
@@ -68,6 +79,10 @@ public static Builder builder(String address) {
/**
* Enqueue a message to the specified queue.
*
+ *