Package eu.vstoyanov.natsy.jetstream

Class StreamContext

  • All Implemented Interfaces:
    
public final class StreamContext

    Context for JetStream stream operations

    Provides access to stream management and operations including:

    • Stream information and updates

    • Message purging and deletion

    • Consumer listing

    Every operation reports failure by throwing a eu.vstoyanov.natsy.exception.NatsyException. The lookups - getMessage, getConsumer - answer null for something that is not there; everything else names a thing it acts on, so a missing one is eu.vstoyanov.natsy.exception.NatsyNotFoundException.

    Example usage:

    val streamCtx = js.stream("MY_STREAM") {
    subjects = listOf("orders.*")
    maxMessages = 1000
}

// Update the stream
streamCtx.update {
    maxMessages = 2000
}

// Purge every message
streamCtx.purge()

// Purge a single subject, or a subtree with "orders.>"
streamCtx.purge("orders.cancelled")

// List consumers
streamCtx.consumers().collect { consumer ->
    println(consumer.name)
}

    • Nested Class Summary

      Nested Classes 
      Modifier and Type Class Description

    • Field Summary

      Fields 
      Modifier and Type Field Description
      private final String name

    • Constructor Summary

      Constructors 
      Constructor Description

    • Enum Constant Summary

      Enum Constants 
      Enum Constant Description

    • Method Summary

      Modifier and Type Method Description
      final String getName() Name of the stream
      final StreamInfo getInfo() Get current stream information
      final StreamInfo update(Function1<StreamConfigBuilder, Unit> block) Update the stream configurationApplies the configuration block to update stream settings.
      final Boolean delete() Delete the streamPermanently deletes the stream and all its messages.
      final PurgeResponse purge(String subject) Purge messages from the streamWithout a subject every message in the stream is removed.
      final Boolean deleteMessage(Long sequence) Delete a specific message by sequence number
      final MessageInfo getMessage(Long sequence) Get a specific message by sequence number
      final Flow<ConsumerInfo> consumers() List all consumers on this streamA failure reaches the collector rather than completing the flow: an empty result means the stream has no consumers, and nothing else.
      final ConsumerContext consumer(String consumerName, Function1<ConsumerConfigBuilder, Unit> block) Create or update a consumer on this streamIf the consumer doesn't exist, a new one is created from the configuration block.
      final ConsumerContext getConsumer(String consumerName) Get an existing consumer on this stream by name

      • Methods inherited from class java.lang.Object

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

    • Constructor Detail

    • Method Detail

      • getInfo

         final StreamInfo getInfo()

        Get current stream information

        Returns:

        the stream's current state as the server holds it

      • update

         final StreamInfo update(Function1<StreamConfigBuilder, Unit> block)

        Update the stream configuration

        Applies the configuration block to update stream 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.

        Parameters:
        block - Configuration block with stream updates
        Returns:

        the updated stream information

      • delete

         final Boolean delete()

        Delete the stream

        Permanently deletes the stream and all its messages. This operation cannot be undone.

        Returns:

        whether the server reported the deletion as done

      • purge

         final PurgeResponse purge(String subject)

        Purge messages from the stream

        Without a subject every message in the stream is removed. With a subject only matching messages are removed, using standard NATS wildcards:

        • * matches exactly one token - orders.* matches orders.created but not orders.created.eu

        • > matches one or more trailing tokens - orders.> matches both

        A subject of > therefore purges the entire stream, same as passing null. A blank subject is rejected rather than honoured: an empty filter is dropped during request serialization, so it would silently widen a targeted purge into a whole-stream wipe.

        Parameters:
        subject - Optional subject filter; when null the entire stream is purged
        Returns:

        purge statistics

      • deleteMessage

         final Boolean deleteMessage(Long sequence)

        Delete a specific message by sequence number

        Parameters:
        sequence - The sequence number of the message to delete
        Returns:

        whether the server reported the deletion as done

      • getMessage

         final MessageInfo getMessage(Long sequence)

        Get a specific message by sequence number

        Parameters:
        sequence - The sequence number of the message to retrieve
        Returns:

        the message, or null if the stream holds no message at that sequence

      • consumers

         final Flow<ConsumerInfo> consumers()

        List all consumers on this stream

        A failure reaches the collector rather than completing the flow: an empty result means the stream has no consumers, and nothing else.

        Returns:

        Flow of ConsumerInfo

      • consumer

         final ConsumerContext consumer(String consumerName, Function1<ConsumerConfigBuilder, Unit> block)

        Create or update a consumer on this stream

        If the consumer doesn't exist, a new one is created from the configuration block. If it already exists, only the properties the block assigns are changed; everything else keeps the value the server currently holds, including settings natsy's DSL does not model.

        Example:

        val consumerCtx = streamCtx.consumer("MY_CONSUMER") {
    durable = "MY_CONSUMER"
    ackPolicy = AckPolicy.EXPLICIT
    maxDeliver = 5
}
        Parameters:
        consumerName - Optional name for the consumer (null for ephemeral)
        block - Configuration block for the consumer
        Returns:

        a context for the consumer that is now in place

      • getConsumer

         final ConsumerContext getConsumer(String consumerName)

        Get an existing consumer on this stream by name

        Parameters:
        consumerName - The name of the consumer
        Returns:

        a context for the consumer, or null if the stream has no consumer by that name