Package eu.vstoyanov.natsy.jetstream

Class JetStreamBatch

  • All Implemented Interfaces:
    
public final class JetStreamBatch<T extends Object>

    Wrapper for a batch of JetStream messages from a pull consumer

    Provides batch-level operations for efficient message processing including batch acknowledgment and selective acknowledgment up to a specific message.

    Example usage:

    js.pull<OrderEvent>(
    stream = "ORDERS",
    consumer = "order-processor",
    batchSize = 100
).collect { batch ->
    // Process batch
    batch.messages.forEach { msg ->
        processOrder(msg.payload)
    }
    // Acknowledge entire batch
    batch.ackAll()

    // Or process with flow operators
    batch.asFlow()
        .filter { it.payload.status == "PENDING" }
        .collect { msg ->
            processOrder(msg.payload)
            msg.ack()
        }
}

    • Nested Class Summary

      Nested Classes 
      Modifier and Type Class Description

    • Constructor Summary

      Constructors 
      Constructor Description

    • Enum Constant Summary

      Enum Constants 
      Enum Constant Description

    • Method Summary

      Modifier and Type Method Description
      final List<JetStreamMessage<T>> getMessages() The list of messages in this batch
      final Boolean getHasMore() Indicates if there are more messages available to fetch
      final Integer getSize() The number of messages in this batch
      final Boolean isEmpty() Check if the batch is empty
      final Boolean isNotEmpty() Check if the batch has messages
      final Unit ackAll() Acknowledge every message in the batchThe whole batch is always attempted: stopping at the first failure would leave the rest unacknowledged and waiting to be redelivered.
      final Unit nakAll(Duration delay) Negatively acknowledge every message in the batch, requesting redeliveryUse this when the whole batch cannot be processed - a database is unavailable, a downstream service is down.
      final Unit ackUpTo(JetStreamMessage<T> message) Acknowledge every message from the start of the batch up to and including messageThe run is taken in delivery order, so acknowledging the fifth message acknowledges the first five.
      final Flow<JetStreamMessage<T>> asFlow() Convert the batch to a Flow of individual messagesThis allows using Flow operators for processing batch messages.
      String toString()

      • Methods inherited from class java.lang.Object

        clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    • Constructor Detail

    • Method Detail

      • getHasMore

         final Boolean getHasMore()

        Indicates if there are more messages available to fetch

      • getSize

         final Integer getSize()

        The number of messages in this batch

      • ackAll

         final Unit ackAll()

        Acknowledge every message in the batch

        The whole batch is always attempted: stopping at the first failure would leave the rest unacknowledged and waiting to be redelivered. If any acknowledgment failed, the first failure is thrown once the loop is done, with the others attached as suppressed exceptions so none of them is lost.

        Non-blocking - JetStreamMessage.ack only enqueues.

      • nakAll

         final Unit nakAll(Duration delay)

        Negatively acknowledge every message in the batch, requesting redelivery

        Use this when the whole batch cannot be processed - a database is unavailable, a downstream service is down. Failure handling is as in ackAll: every message is attempted, and the first failure is thrown afterwards carrying the rest.

        Non-blocking - JetStreamMessage.nak only enqueues.

        Parameters:
        delay - Optional delay before redelivery (null for immediate redelivery)

      • ackUpTo

         final Unit ackUpTo(JetStreamMessage<T> message)

        Acknowledge every message from the start of the batch up to and including message

        The run is taken in delivery order, so acknowledging the fifth message acknowledges the first five. This is the operation for a handler that processed a prefix of the batch and stopped: the run it got through is acknowledged, and the rest is redelivered.

        How many acknowledgements that takes is the consumer's business rather than the caller's, and it follows the policy the server reports for this consumer, never the one a configuration block asked for. Under AckPolicy.ALL one acknowledgement closes the run - "all messages with a sequence number less than the message acked are also acknowledged" (jnats io/nats/client/api/AckPolicy.java:30) - so only message is sent. Under AckPolicy.EXPLICIT, natsy's default, "each message must be acknowledged individually" (:34), so every message in the run is.

        The whole run is always attempted, exactly as in ackAll: if any acknowledgement failed the first is thrown once the run is done, the others attached as suppressed exceptions.

        message must be from this batch - passing one that is not is a programming error, not a NATS failure, so it fails with IllegalArgumentException.

        Unlike the rest of the acknowledgement surface this one blocks: message's acknowledgement goes out with ackSync, which waits for the server to confirm it. The acknowledgements ahead of it are only enqueued, as ackAll's are, but they travel the same connection first.

        Parameters:
        message - The message up to which to acknowledge (inclusive)

      • asFlow

         final Flow<JetStreamMessage<T>> asFlow()

        Convert the batch to a Flow of individual messages

        This allows using Flow operators for processing batch messages.

        Example:

        batch.asFlow()
    .filter { it.payload.amount > 100 }
    .collect { msg ->
        processHighValueOrder(msg.payload)
        msg.ack()
    }
        Returns:

        Flow of JetStreamMessage<T>