Constructor Detail

• NatsKlient

  NatsKlient(KlientConfig config, CoroutineContext parentContext)

  Parameters:

  config - the validated configuration, from KlientConfigBuilder.build

  parentContext - context the klient's own background work runs in - a Ktor application.coroutineContext, say, so that work inherits the host's dispatcher and whatever else it carries.

• NatsKlient

  NatsKlient(CoroutineContext parentContext, Function1<KlientConfigBuilder, Unit> configure)

  Build and configure in one step: NatsKlient { servers = listOf(...) }

Method Detail

• getJetStream

   final JetStreamContext getJetStream()

  JetStream, key-value and object-store operations

  Created on first access and reused. Building it does no I/O - the jnats contexts behind it are themselves lazy - so reaching for it before connect is harmless.

• getConnection

   final Connection getConnection()

  Direct access to the underlying NATS connection for advanced use cases

  One jnats behaviour differs from its default here: natsy builds the connection with advancedRequestBehavior(), which is what lets it tell "nobody is subscribed" apart from "nobody answered in time". A Connection.request made directly on this connection therefore returns an io.nats.client.RequestFailureMessage carrying the reason where a stock jnats connection would return null.

• getStatus

   final ConnectionStatus getStatus()

  What the connection is currently doing

  Unlike connection this never throws: before connect it reports ConnectionStatus.DISCONNECTED. That makes it usable from a health or readiness probe that may run before, during or after the connection's life - which is also why it is natsy's own enum, so writing that probe needs no jnats import.

• getStatistics

   final Statistics getStatistics()

  Access to connection statistics

• connect

   final NatsKlient connect()

  Open the connection, or return immediately if it is already open

  Synchronous and fail-fast: jnats connects before it returns, or it throws. Idempotent, so a container that calls it on every startup path only dials once.

  Blocks the calling thread for the handshake, which is what a startup hook wants; from a coroutine use connectSuspending.

  Returns:

  this klient, so NatsKlient { }.connect() reads as one expression

• connectSuspending

   final NatsKlient connectSuspending()

  connect for callers that can suspend

  Connecting parks a thread for the handshake, so this runs it on the configured KlientConfigBuilder.ioDispatcher rather than on the caller's.

• publishTyped

   <T extends Any> Unit publishTyped(String subject, T message, KType type, Headers headers)

  publish with the type to carry message as given as a value.

• publishBytes

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

  Publish raw bytes for performance-critical paths

  Hands the message to the connection's outgoing queue and returns without waiting for the server. Thread-safe, and needs no dispatcher of its own.

  It is not unconditionally non-blocking, though. The outgoing queue is bounded - KlientConfigBuilder.maxMessagesInOutgoingQueue, 5000 by default - and a publisher faster than the connection can write will fill it. jnats then waits for room, and gives up with a eu.vstoyanov.natsy.exception.NatsyConnectionException rather than blocking indefinitely, so under sustained backpressure a publish can park the calling thread for seconds. Set KlientConfigBuilder.discardMessagesWhenOutgoingQueueFull to drop the message instead of waiting.

• requestTyped

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

  request with both types given as values.

• requestManyTyped

   <Req extends Any, Resp extends Any> Flow<Resp> requestManyTyped(String subject, Req request, KType requestType, KType responseType, Integer expectedResponses, Duration timeout, DecodeErrorStrategy onDecodeError)

  requestMany with both types given as values.

• subscribeTyped

   <T extends Any> SubscriptionHandle<T> subscribeTyped(String subject, KType type, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError)

  subscribe with the type to decode into given as a value.

• subscribeMessagesTyped

   <T extends Any> SubscriptionHandle<TypedMessage<T>> subscribeMessagesTyped(String subject, KType type, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError)

  subscribeMessages with the type to decode into given as a value.

• respondTyped

   <Req extends Any, Resp extends Any> SubscriptionHandle<TypedMessage<Req>> respondTyped(String subject, KType requestType, KType responseType, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError, SuspendFunction1<Req, Resp> handler)

  respond with both types given as values.

  Left to the implementation, unlike the shape of the flow it returns, which is fixed: subscribeMessagesTyped with the reply folded in. What an implementation adds is the response codec, resolved before the handle exists - only something holding a registry can do that, and respond's @throws promises it.

• subscribeBytes

   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.

  The flow ends when its collector is cancelled, when NatsKlient.close is called, and when the connection itself reaches CLOSED - a NatsKlient.drain, or jnats giving up reconnecting - so a subscription collected for the klient's whole lifetime completes rather than hanging on a connection that is no longer there.

  Example:

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

  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

• drain

   final Boolean drain(Duration timeout)

  Gracefully drain the connection, then close it

  Suspends until the drain has actually finished. jnats only unsubscribes and flushes before its Connection.drain returns - delivering the remaining in-flight messages, blocking further publishing, doing a last flush and closing the connection all happen on a background thread, and only the returned future tells you when that is done. This awaits that future, so once this returns it is safe to exit the process without losing messages.

  Draining closes the connection, so the klient is not reusable afterwards. The connection reaching CLOSED is also what ends every live flow this klient handed out - a collector waiting on a drain does not need a subsequent close to complete. close is still worth calling to release the scope-side resources; it returns immediately on an already-closed connection. closeGracefully is that pair in one call.

  Parameters:

  timeout - budget for the whole drain.

  Returns:

  true if every subscription drained within timeout, false if the timeout forced the close. Also true when the connection was already closed and there was nothing to do.

• closeGracefully

   final Boolean closeGracefully(Duration timeout)

  Drain, then close: the complete shutdown sequence in one call

  drain delivers what is already in flight and closes the connection; close then ends the klient's flows and cancels its background work. The order matters - closing first would kill the subscriptions the drain exists to deliver to.

  The klient always ends up closed, even if the drain fails or the caller is cancelled mid-way.

  Drain waits for delivery, not for your handlers: a collector still processing its last message when the drain finishes is cancelled like any other. Work that has to finish belongs in front of this call.

  Parameters:

  timeout - budget for the drain, as in drain

  Returns:

  true if every subscription drained within timeout, false if the timeout forced the close

• closeBlocking

   final Boolean closeBlocking(Duration timeout)

  closeGracefully for callers that cannot suspend

  Blocks the calling thread for up to timeout on the drain. That is what non-suspending shutdown hooks need - Koin's onClose, a JVM shutdown hook, a Ktor MonitoringEvent handler - where the process must not exit until the drain has finished.

  From inside a coroutine closeGracefully is the one to call: this parks the thread it runs on for the whole drain.

  Parameters:

  timeout - budget for the drain, as in drain

  Returns:

  true if every subscription drained within timeout, false if the timeout forced the close

• close

   Unit close()

  End everything this klient is running and close the connection immediately

  closeGracefully is the graceful counterpart: it drains first, so nothing in flight is lost.

  Every subscription, watch and consumer flow the klient handed out is ended, so its collector completes normally rather than waiting on a channel nothing will ever close again. Then the connection closes, and the klient's background work is cancelled.

  Cancellation is signalled, not awaited: close cannot suspend, so a coroutine that is still unwinding may briefly outlive the call. For the same reason a flow's own cleanup - deleting an ephemeral consumer, say - runs on a best-effort basis; the server reaps what is left when the connection drops.

  Blocks briefly while jnats joins its reader and writer threads. Idempotent, and safe on a klient that was never connected.