Constructor Detail

Method Detail

• getCoroutineContext

   CoroutineContext getCoroutineContext()

• getStatistics

   final Statistics getStatistics()

• publish

   final <T extends Any> Unit publish(String subject, T message, KClass<T> type, Headers headers)

  Publish a typed message with automatic serialization

  This operation is non-blocking and thread-safe. The NATS client handles buffering internally, so no I/O dispatcher is needed.

  Parameters:

  subject - The NATS subject to publish to

  message - The message to publish

  type - The Kotlin class of the message type

  headers - Optional headers to include

• publish

   final <T extends Any> Unit publish(String subject, T message, Headers headers)

  Publish with reified type parameter for better ergonomics

  This operation is non-blocking and thread-safe.

  Parameters:

  subject - The NATS subject to publish to

  message - The message to publish

  headers - Optional headers to include

• publishBytes

   final Unit publishBytes(String subject, ByteArray data, Headers headers)

  Publish raw bytes for performance-critical paths

  This operation is non-blocking and thread-safe. The NATS Java client's connection.publish() method is thread-safe and only adds to an internal buffer, so no I/O dispatcher or suspension is needed.

  Parameters:

  subject - The NATS subject to publish to

  data - The raw byte data to publish

  headers - Optional headers to include

• request

   final <Req extends Any, Resp extends Any> Resp request(String subject, Req request, KClass<Req> requestType, KClass<Resp> responseType, Duration timeout)

  Request-reply pattern with typed messages

  Uses Dispatchers.IO because connection.request() blocks waiting for response. This is a synchronous blocking operation that must not run on the default dispatcher.

  Parameters:

  subject - The NATS subject to send request to

  request - The request message

  requestType - The Kotlin class of the request type

  responseType - The Kotlin class of the response type

  timeout - Maximum time to wait for response

  Returns:

  the decoded response

• request

   final <Req extends Any, Resp extends Any> Resp request(String subject, Req request, Duration timeout)

  Request-reply with reified type parameters for better ergonomics

• requestMany

   final <Req extends Any, Resp extends Any> Flow<Resp> requestMany(String subject, Req request, KClass<Req> requestType, KClass<Resp> responseType, Integer expectedResponses, Duration timeout)

  Request-many pattern (scatter-gather)

  Sends a request to multiple responders and collects all responses until timeout or expectedResponses count is reached.

  The response channel is unbounded — the volume is already capped by expectedResponses and timeout — so no response is ever dropped on its way to the collector, and expectedResponses counts responses that actually reached it.

• requestMany

   final <Req extends Any, Resp extends Any> Flow<Resp> requestMany(String subject, Req request, Integer expectedResponses, Duration timeout)

  Request-many with reified type parameters for better ergonomics

• subscribe

   final <T extends Any> SubscriptionHandle<T> subscribe(String subject, KClass<T> type, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError)

  Subscribe to a subject with typed messages

  Messages are delivered through a channel of capacity slots and decoded downstream of it. With the default Channel.UNLIMITED no message is ever dropped; see subscribeBytes for the buffering contract.

  Decoding sits between the delivery channel and the collector, which breaks Flow operator fusion: a .buffer() applied to SubscriptionHandle.messages allocates a second buffer and leaves the delivery channel untouched. Size the buffer that actually holds incoming messages with capacity and onOverflow.

  Example:

  val sensors = subscribe<Telemetry>("sensors", capacity = 128, onOverflow = BufferOverflow.DROP_OLDEST)
  launch { sensors.messages.collect { telemetry -> ... } }
  sensors.started()

  Parameters:

  subject - The NATS subject to subscribe to

  type - The Kotlin class of the expected message type

  queue - Optional queue group name for load balancing

  capacity - Delivery channel capacity, Channel.UNLIMITED by default

  onOverflow - What to do when a bounded delivery channel is full

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

  Returns:

  a SubscriptionHandle whose SubscriptionHandle.messages carries decoded values

• subscribe

   final <T extends Any> SubscriptionHandle<T> subscribe(String subject, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError)

  Subscribe with reified type parameter for better ergonomics

  Messages are delivered through a channel of capacity slots and decoded downstream of it. With the default Channel.UNLIMITED no message is ever dropped. A .buffer() applied to SubscriptionHandle.messages cannot resize that channel — decoding breaks operator fusion — so use capacity and onOverflow instead.

  Example:

  val sensors = subscribe<Telemetry>("sensors", capacity = 128, onOverflow = BufferOverflow.DROP_OLDEST)
  launch { sensors.messages.collect { telemetry -> ... } }
  sensors.started()

  Parameters:

  subject - The NATS subject to subscribe to

  queue - Optional queue group name for load balancing

  capacity - Delivery channel capacity, Channel.UNLIMITED by default

  onOverflow - What to do when a bounded delivery channel is full

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

• subscribeBytes

   final SubscriptionHandle<Message> subscribeBytes(String subject, String queue, Integer capacity, BufferOverflow onOverflow)

  Subscribe to raw messages for performance

  The jnats dispatcher hands messages to a channel of capacity slots without ever suspending. The default Channel.UNLIMITED mirrors the unbounded queue jnats keeps for the dispatcher itself, so no message is ever dropped — at the cost of unbounded memory growth if the collector cannot keep up.

  Bound the buffer with capacity to trade delivery for memory. Because the dispatcher callback cannot suspend, BufferOverflow.SUSPEND on a full bounded buffer means the message is dropped and logged at WARN; BufferOverflow.DROP_OLDEST and BufferOverflow.DROP_LATEST drop silently, by definition.

  The buffer is applied inside the flow, so a .buffer() on SubscriptionHandle.messages fuses into the same channel instead of adding a second one: a non-SUSPEND overflow policy replaces both settings, while a plain .buffer(n) cannot shrink an unbounded buffer.

  Example:

  val telemetry = subscribeBytes("telemetry", capacity = 128, onOverflow = BufferOverflow.DROP_OLDEST)
  launch { telemetry.messages.collect { msg -> ... } }
  telemetry.started()

  Parameters:

  subject - The NATS subject to subscribe to

  queue - Optional queue group name for load balancing

  capacity - Delivery channel capacity, Channel.UNLIMITED by default

  onOverflow - What to do when a bounded delivery channel is full

  Returns:

  a SubscriptionHandle; SubscriptionHandle.started is the readiness rendezvous

• flush

   final Unit flush(Duration timeout)

  Flushes every operation the connection has already been asked to send, and waits for the server to acknowledge it - jnats' Connection.flush.

  Not a readiness primitive: it can only order against work already issued, and a cold subscription is issued when its flow is collected. Use SubscriptionHandle.started.

  Parameters:

  timeout - How long to wait for the server acknowledgment

• headers

   final Headers headers(Function1<Headers, Unit> block)

  Create headers using DSL