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

    • 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 all messages up to and including the specified messageThis is an efficient operation that tells JetStream to acknowledge all messages up to the specified message in a single operation.
      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

    • 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 all messages up to and including the specified message

        This is an efficient operation that tells JetStream to acknowledge all messages up to the specified message in a single operation. This is useful for processing messages in order and acknowledging progress.

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

        Note: This operation acknowledges messages by sequence, which is more efficient than acknowledging each message individually when processing in order.

        Unlike the rest of the acknowledgement surface this one blocks: ackSync waits for the server to confirm.

        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>