Constructor Detail

Method Detail

• getManagement

   final JetStreamManagement getManagement()

  Access to the underlying JetStream management client for administrative operations.

  Use this for stream and consumer lifecycle management:

  • Creating, updating, and deleting streams

  • Creating, updating, and deleting consumers

  • Retrieving stream and consumer information

  • Purging streams and managing messages

  For publishing and consuming messages, use the publish() and subscribe() methods on JetStreamContext directly rather than accessing the management API.

  Returns:

  The JetStreamManagement client (thread-safe)

• publish

   final <T extends Any> PublishAck publish(String subject, T message, Headers headers, PublishExpectations expectations)

  Publish a typed message to JetStream with optional expectations

  Uses Dispatchers.IO because JetStream publish operations block waiting for PublishAck from the server. Unlike core NATS publish (which just buffers), JetStream publish is a synchronous API call that must not run on the default dispatcher.

  The codec's content type is always stamped onto the outgoing headers, and headers and expectations combine freely: jnats merges the expectation headers into the caller's own (JetStream.publish(subject, headers, body, options)).

  Parameters:

  subject - The subject to publish to

  message - The message to publish

  headers - Optional headers to include

  expectations - Optional publish expectations for ordering and deduplication

  Returns:

  the server's acknowledgement, naming the stream and sequence it was stored at

• publishBytes

   final PublishAck publishBytes(String subject, ByteArray data, Headers headers, PublishExpectations expectations)

  Publish raw bytes to JetStream

  Uses Dispatchers.IO because JetStream publish operations block waiting for PublishAck from the server. This is a synchronous API call that must not run on the default dispatcher.

  No content type is stamped - raw bytes carry no codec - but headers and expectations combine freely: jnats merges the expectation headers into the caller's own (JetStream.publish(subject, headers, body, options)).

  Parameters:

  subject - The subject to publish to

  data - The raw data to publish

  headers - Optional headers to include

  expectations - Optional publish expectations

  Returns:

  the server's acknowledgement, naming the stream and sequence it was stored at

• stream

   final StreamContext stream(String name, Function1<StreamConfigBuilder, Unit> block)

  Create or update a JetStream stream

  If the stream 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 streamCtx = js.stream("MY_STREAM") {
      subjects = listOf("orders.*", "payments.*")
      retentionPolicy = RetentionPolicy.LIMITS
      maxMessages = 1000000
      storageType = StorageType.FILE
      replicas = 3
  }

  Parameters:

  name - The name of the stream

  block - Configuration block for the stream

  Returns:

  a context for the stream that is now in place

• getStream

   final StreamContext getStream(String name)

  Get an existing stream by name

  Parameters:

  name - The name of the stream to retrieve

  Returns:

  a context for the stream, or null if there is no stream by that name

• streams

   final Flow<StreamContext> streams()

  List all streams in the JetStream account

  Returns a Flow that emits StreamContext for each stream.

  Example:

  js.streams().collect { streamCtx ->
      val info = streamCtx.getInfo()
      println("Stream: ${info.configuration.name}")
  }

  Returns:

  Flow of StreamContext

• deleteStream

   final Boolean deleteStream(String name)

  Delete a stream by name (convenience method)

  Parameters:

  name - The name of the stream to delete

  Returns:

  whether the server reported the deletion as done

• consumer

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

  Create or update a JetStream consumer

  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.

  For ephemeral consumers, set name to null and don't set the durable property. An auto-generated name will be used.

  Example:

  // Durable consumer
  val consumerCtx = js.consumer("MY_STREAM", "MY_CONSUMER") {
      durable = "MY_CONSUMER"
      ackPolicy = AckPolicy.EXPLICIT
      maxDeliver = 5
      filterSubject = "orders.>"
  }
  
  // Ephemeral consumer
  val ephemeralCtx = js.consumer("MY_STREAM", null) {
      ackPolicy = AckPolicy.EXPLICIT
      filterSubject = "orders.pending"
  }

  Parameters:

  stream - The name of the stream

  name - 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 stream, String name)

  Get an existing consumer by name

  Parameters:

  stream - The name of the stream

  name - The name of the consumer

  Returns:

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

• deleteConsumer

   final Boolean deleteConsumer(String stream, String name)

  Delete a consumer by name (convenience method)

  Parameters:

  stream - The name of the stream

  name - The name of the consumer to delete

  Returns:

  whether the server reported the deletion as done

• subscribe

   final <T extends Any> Flow<JetStreamMessage<T>> subscribe(String stream, String filterSubject, String queue, String consumerName, Function1<ConsumerConfigBuilder, Unit> config, Integer bufferSize, DecodeErrorStrategy onDecodeError)

  Subscribe to a JetStream stream with a push consumer

  Creates an ephemeral push consumer and returns a Flow of typed messages. Each message includes acknowledgment controls and metadata.

  The consumer will automatically create a unique deliver subject and subscribe to it. Messages are decoded using the codec registry and wrapped in JetStreamMessage for type-safe processing with acknowledgment controls.

  Flow Control: When enabled via ConsumerConfig.flowControl, the subscription binds through jnats' push API, so its PushMessageManager answers the server's flow control requests and absorbs idle heartbeats before they reach this flow. No additional code is required. Flow control and idle heartbeat cannot be combined with queue: queue members share a deliver subject, so only one of them would ever see - and answer - a flow control request.

  Example:

  js.subscribe<OrderEvent>(
      stream = "ORDERS",
      filterSubject = "orders.pending"
  ).collect { msg ->
      processOrder(msg.payload)
      msg.ack()
  }

  Queue groups (work distribution) require consumerName: members balance by subscribing to one shared consumer's deliver subject, and consumerName is what makes every member bind to that same consumer. Passing queue without it throws IllegalArgumentException - it would give each subscriber its own ephemeral consumer, and therefore a full copy of the stream rather than a share of the work.

  js.subscribe<OrderEvent>(
      stream = "ORDERS",
      queue = "order-workers",
      consumerName = "order-worker",   // shared by every member of the group
      filterSubject = "orders.>"
  ).collect { msg ->
      // Work is distributed across multiple subscribers in the queue group
      processOrder(msg.payload)
      msg.ack()
  }

  Parameters:

  stream - The name of the stream to subscribe to

  filterSubject - Optional subject filter (e.g., "orders.

  queue - Optional queue group name for work distribution; requires consumerName

  consumerName - Optional durable consumer name (null for ephemeral); mandatory when queue is set

  config - Configuration block for consumer settings

  bufferSize - Delivery channel capacity.

  onDecodeError - What to do with a message that cannot be decoded, see DecodeErrorStrategy.

  Returns:

  Flow of JetStreamMessage<T> with acknowledgment controls

• pull

   final <T extends Any> Flow<JetStreamBatch<T>> pull(String stream, String consumer, Integer batchSize, Duration timeout, Boolean noWait, Function1<ConsumerConfigBuilder, Unit> config, DecodeErrorStrategy onDecodeError)

  Create a pull consumer that fetches messages in batches

  Creates a durable pull consumer and returns a Flow of message batches. Each batch contains the requested number of messages (or fewer if not available). This is the most efficient way to consume large volumes of messages.

  The consumer must be durable (have a name) since pull consumers require explicit management. Use the batch operations to efficiently acknowledge all messages at once or selectively acknowledge up to a specific message.

  Example:

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

  For flow-based processing with operators:

  js.pull<OrderEvent>("ORDERS", "processor", 50)
      .collect { batch ->
          batch.asFlow()
              .filter { it.payload.amount > 100 }
              .collect { msg ->
                  processHighValueOrder(msg.payload)
                  msg.ack()
              }
      }

  Parameters:

  stream - The name of the stream to pull from

  consumer - The name of the durable pull consumer

  batchSize - Number of messages to fetch per batch (default: 10); rejected if the consumer caps batches below it via ConsumerConfigBuilder.maxBatch

  timeout - Maximum time to wait for messages in each fetch (default: 1 second)

  noWait - If true, return immediately with available messages without waiting for full batch (default: false)

  config - Optional configuration block for consumer settings (only used if consumer doesn't exist)

  onDecodeError - What to do with a message that cannot be decoded, see DecodeErrorStrategy.

  Returns:

  Flow of JetStreamBatch<T> with batch operations

• keyValue

   final KeyValueStore keyValue(String bucket, Function1<KeyValueConfigBuilder, Unit> block)

  Create or access a Key-Value store bucket

  If the bucket already exists, it will be returned. If it doesn't exist, a new bucket will be created with the specified configuration.

  Example:

  val kv = js.keyValue("user-preferences") {
      description = "User preference storage"
      maxHistoryPerKey = 5
      ttl = 24.hours
      storageType = StorageType.MEMORY
  }
  
  kv.put("user:123:theme", "dark")
  val theme = kv.get<String>("user:123:theme") ?: "light"

  Parameters:

  bucket - The name of the bucket

  block - Configuration block for the bucket (optional)

  Returns:

  the bucket, created if it was not already there

• getKeyValue

   final KeyValueStore getKeyValue(String bucket)

  Get an existing Key-Value store bucket

  Parameters:

  bucket - The name of the bucket to retrieve

  Returns:

  the bucket, or null if there is no bucket by that name

• deleteKeyValue

   final Boolean deleteKeyValue(String bucket)

  Delete a Key-Value store bucket (convenience method)

  Parameters:

  bucket - The name of the bucket to delete

  Returns:

  true once the bucket is gone

• objectStore

   final ObjectStore objectStore(String bucket, Function1<ObjectStoreConfigBuilder, Unit> block)

  Create or access an Object Store bucket

  If the bucket already exists, it will be returned. If it doesn't exist, a new bucket will be created with the specified configuration.

  Object stores provide efficient storage for large files and binary data with streaming support to avoid loading entire objects into memory.

  Example:

  val os = js.objectStore("file-storage") {
      description = "Application file storage"
      storageType = StorageType.FILE
      maxBucketSize = 10 * 1024 * 1024 * 1024L // 10GB
      replicas = 3
  }
  
  // Upload a file
  os.putFile("backup.zip", File("/path/to/backup.zip"))
  
  // Download a file
  os.getFile("backup.zip", File("/path/to/restore.zip"))

  Parameters:

  bucket - The name of the bucket

  block - Configuration block for the bucket (optional)

  Returns:

  the bucket, created if it was not already there

• getObjectStore

   final ObjectStore getObjectStore(String bucket)

  Get an existing Object Store bucket

  Parameters:

  bucket - The name of the bucket to retrieve

  Returns:

  the bucket, or null if there is no bucket by that name

• deleteObjectStore

   final Boolean deleteObjectStore(String bucket)

  Delete an Object Store bucket (convenience method)

  Parameters:

  bucket - The name of the bucket to delete

  Returns:

  true once the bucket is gone