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().getOrThrow()

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

    • 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 Result<Unit> ackAll() Acknowledge all messages in the batchIterates through all messages and acknowledges each one.
      final Result<Unit> nakAll(Duration delay) Negatively acknowledge all messages in the batchIterates through all messages and negatively acknowledges each one with an optional delay before redelivery.
      final Result<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 Result<Unit> ackAll()

        Acknowledge all messages in the batch

        Iterates through all messages and acknowledges each one. If any acknowledgment fails, the operation continues but returns a failure result with the first error. This allows partial batch processing where some messages may have already been successfully acknowledged.

        This operation is non-blocking since message.ack() is non-blocking. No I/O dispatcher needed.

        Returns:

        Result indicating success or failure of the batch acknowledgment

      • nakAll

         final Result<Unit> nakAll(Duration delay)

        Negatively acknowledge all messages in the batch

        Iterates through all messages and negatively acknowledges each one with an optional delay before redelivery. If any NAK fails, the operation continues but returns a failure result with the first error.

        Use this when the entire batch cannot be processed (e.g., database unavailable).

        This operation is non-blocking since message.nak() is non-blocking. No I/O dispatcher needed.

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

        Result indicating success or failure of the batch NAK

      • ackUpTo

         final Result<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.

        The message must be from this batch. If the message is not found in the batch, an error is returned.

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

        IMPORTANT: This operation uses ackSync() which IS a blocking operation that waits for server confirmation. Therefore, Dispatchers.IO is required.

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

        Result indicating success or failure of the selective acknowledgment

      • 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().getOrThrow()
    }
        Returns:

        Flow of JetStreamMessage<T>