Package eu.vstoyanov.natsy.jetstream

Class ConsumerContext

  • All Implemented Interfaces:
    
public final class ConsumerContext

    Context for JetStream consumer operations

    Provides access to consumer management and monitoring including:

    • Consumer information and updates

    • Pause/resume operations

    • Monitoring metrics

    Example usage:

    val consumerCtx = js.consumer("MY_STREAM", "MY_CONSUMER") {
    durable = "MY_CONSUMER"
    ackPolicy = AckPolicy.EXPLICIT
    maxDeliver = 5
}

// Update the consumer
consumerCtx.update {
    maxDeliver = 10
}

// Pause the consumer until it is explicitly resumed
consumerCtx.pause()

// ...or until a deadline, after which the server resumes it automatically
consumerCtx.pause(30.minutes)

// Everything worth monitoring, in one round trip
val stats = consumerCtx.stats()
println("${stats.ackPending} unacknowledged, ${stats.pending} still to come")

    • 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 String getStreamName() Name of the stream this consumer belongs to
      final String getConsumerName() Name of the consumer
      final ConsumerInfo getInfo() Get current consumer information, exactly as jnats models itThe escape hatch for anything stats does not carry.
      final ConsumerStats stats() Everything worth monitoring about this consumer, in one round tripReplaces the six single-value monitors this class used to expose.
      final ConsumerInfo update(Function1<ConsumerConfigBuilder, Unit> block) Update the consumer configurationApplies the configuration block to update consumer settings.
      final Boolean delete() Delete the consumerPermanently deletes the consumer.
      final PauseResponse pause(Instant until) Pause the consumerTemporarily pauses message delivery to the consumer.
      final PauseResponse pause(Duration duration) Pause the consumer for a durationConvenience over pause with an absolute deadline.
      final Boolean resume() Resume the consumerResumes message delivery to a paused consumer.

      • Methods inherited from class java.lang.Object

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

    • Constructor Detail

    • Method Detail

      • getInfo

         final ConsumerInfo getInfo()

        Get current consumer information, exactly as jnats models it

        The escape hatch for anything stats does not carry. stats is the one to reach for otherwise: it is typed to natsy's model and answers the questions a monitor actually asks.

        Returns:

        the consumer's state as the server holds it

      • stats

         final ConsumerStats stats()

        Everything worth monitoring about this consumer, in one round trip

        Replaces the six single-value monitors this class used to expose. Each of those made its own server call and answered false or 0 when that call failed, so a readiness probe read a dead connection as "healthy, nothing pending". This one either answers or throws.

        Returns:

        a snapshot of the consumer's state

      • update

         final ConsumerInfo update(Function1<ConsumerConfigBuilder, Unit> block)

        Update the consumer configuration

        Applies the configuration block to update consumer settings. Only the properties assigned in the block are changed; everything else keeps the value the server currently holds, including settings natsy's DSL does not model.

        Note: Not all properties can be updated after creation. Attempting to change immutable properties will result in an error from the NATS server.

        Parameters:
        block - Configuration block with consumer updates
        Returns:

        the updated consumer information

      • delete

         final Boolean delete()

        Delete the consumer

        Permanently deletes the consumer. This operation cannot be undone. After deletion, the consumer context should not be used.

        Returns:

        whether the server reported the deletion as done

      • pause

         final PauseResponse pause(Instant until)

        Pause the consumer

        Temporarily pauses message delivery to the consumer. Messages will continue to accumulate but won't be delivered until the deadline passes or resume is called.

        NATS models pausing purely as a deadline: a pause_until that is absent or already in the past is a resume request on the same endpoint. Passing a non-future until is therefore rejected instead of silently resuming, and "pause until further notice" (the default) is expressed as a far-future deadline that resume clears immediately.

        Parameters:
        until - when the consumer should automatically resume; must be in the future.
        Returns:

        the pause the server put in place

      • pause

         final PauseResponse pause(Duration duration)

        Pause the consumer for a duration

        Convenience over pause with an absolute deadline. Duration.INFINITE pauses until resume is called.

        Parameters:
        duration - how long to pause for; must be positive
        Returns:

        the pause the server put in place

      • resume

         final Boolean resume()

        Resume the consumer

        Resumes message delivery to a paused consumer.

        Returns:

        whether the server reported the consumer as resumed