Constructor Detail

Method Detail

• publish

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

  Publish a typed message, serializing it with the codec for T

  Any @Serializable type works with no registration; see eu.vstoyanov.natsy.codec.CodecRegistry for what else does and how to override it.

  Hands the message to the outgoing queue and returns; see NatsOperations.publishBytes for what happens when that queue is full.

  Parameters:

  subject - The NATS subject to publish to

  message - The message to publish

  headers - Optional headers to include

• request

   final static <Req extends Any, Resp extends Any> Resp request(NatsOperations $self, String subject, Req request, Duration timeout)

  Request-reply pattern with typed messages

  Parameters:

  subject - The NATS subject to send request to

  request - The request message

  timeout - Maximum time to wait for response

  Returns:

  the decoded response

• requestMany

   final static <Req extends Any, Resp extends Any> Flow<Resp> requestMany(NatsOperations $self, String subject, Req request, Integer expectedResponses, Duration timeout, DecodeErrorStrategy onDecodeError)

  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.

  A reply that cannot be decoded is dropped with a warning by default: one responder speaking a stale dialect is what a scatter-gather round is expected to survive. onDecodeError is where that choice is made, in the same terms subscribe makes it.

  Parameters:

  subject - The NATS subject to send the request to

  request - The request message

  expectedResponses - Stop collecting once this many responses have arrived

  timeout - How long to keep collecting

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

• subscribe

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

  Subscribe to a subject with typed messages

  Any @Serializable type works with no registration; see eu.vstoyanov.natsy.codec.CodecRegistry for what else does and how to override it.

  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.

  Only the payload reaches the collector. subscribeMessages is the same subscription carrying the subject, reply subject and headers alongside it.

  Example:

  val sensors = klient.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: positive, Channel.CONFLATED or Channel.BUFFERED

  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

• subscribeMessages

   final static <T extends Any> SubscriptionHandle<TypedMessage<T>> subscribeMessages(NatsOperations $self, String subject, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError)

  Subscribe to a subject with typed messages, keeping the envelope

  subscribe with what it leaves out: each message arrives as a TypedMessage carrying the subject it was published to, the subject a reply is expected on, and the headers it came with. Everything else - codec resolution, buffering, decode failures - behaves exactly as it does there.

  Worth reaching for when the payload is not the whole message: a wildcard subscription that has to know which subject matched, a handler that reads a trace id off the headers, a responder that has to answer TypedMessage.replyTo - the last of which is respond, written for you.

  Example:

  val events = klient.subscribeMessages<Event>("events.*")
  launch { events.messages.collect { println("${it.subject}: ${it.payload}") } }
  events.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: positive, Channel.CONFLATED or Channel.BUFFERED

  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 messages

• respond

   final static <Req extends Any, Resp extends Any> SubscriptionHandle<TypedMessage<Req>> respond(NatsOperations $self, String subject, String queue, Integer capacity, BufferOverflow onOverflow, DecodeErrorStrategy onDecodeError, SuspendFunction1<Req, Resp> handler)

  Answer typed requests on subject

  The other half of request: decode the request, compute a response, publish it to the subject the caller is listening on. Written on top of subscribeMessages - it is that subscription with the reply folded in - so everything true of a subscription is true here. It is cold, it starts when SubscriptionHandle.messages is collected, SubscriptionHandle.started is the rendezvous that says it is live, and cancelling the collector ends it.

  val pricing = klient.respond<PriceQuery, PriceQuote>("pricing.quote", queue = "pricing") { query ->
      PriceQuote(query.sku, catalogue.priceOf(query.sku))
  }
  val serving = launch { pricing.messages.collect() }
  pricing.started()

  Every element the flow emits is a request that has been answered - which is what makes counting them, logging them or bounding them with take possible. Collecting is also what drives the loop: a handle nobody collects answers nothing.

  One request at a time. The handler runs in the collector's coroutine, so a slow handler is backpressure on the subscription rather than an unbounded pile of concurrent work; run several handles in one queue group when that is not what you want, which is also how a responder scales across processes.

  A message with no reply subject was published rather than requested. It is logged and skipped without running the handler. An exception from the handler fails the flow the way one from any collector's body would, so a service that answers failures rather than dying on them catches inside the handler and returns whatever its protocol says a failure looks like.

  Parameters:

  subject - The NATS subject to answer requests on

  queue - Optional queue group name, so one member of the group answers each request

  capacity - Delivery channel capacity, Channel.UNLIMITED by default: positive, Channel.CONFLATED or Channel.BUFFERED

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

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

  handler - computes the response for a decoded request

  Returns:

  a SubscriptionHandle emitting each request as it is answered