Package eu.vstoyanov.natsy

Interface NatsOperations

  • All Implemented Interfaces:
    
public interface NatsOperations

    Everything that publishes, requests or subscribes - the messaging surface, without the connection behind it.

    NatsKlient implements it, and application code is usually better off naming this. A service that takes a NatsOperations says what it needs - to send and receive messages - rather than that it holds a connection it could close; and a dependency-injection container can bind the two separately, so only the code that owns the lifecycle can end it:

    single(createdAtStart = true) { NatsKlient(configure = configure).connect() }
    .onClose { it?.closeBlocking() }
single<NatsOperations> { get<NatsKlient>() }

    Connecting, draining, closing and JetStream stay on NatsKlient: those belong to whoever owns the connection, not to everyone who sends a message over it.

    Each typed operation appears twice. The one to call is the reified extension - publish("orders", event) - which resolves the codec from the type at the call site; being an extension, it has to be imported. Every one of them delegates to the member named after it with a Typed suffix, which takes the type as a KType value instead. That indirection is what an interface forces: a reified type parameter needs an inline function, and an interface member cannot be inlined.

    The Typed form is worth calling directly when the type arrives as a value rather than as a type argument - a listener container told at startup what to carry. Nothing checks that the KType describes T, though: a mismatch is not a compile error and not a failure at the call either, but a ClassCastException wherever the decoded value is finally used.

    • Constructor Detail

    • Method Detail

      • publishBytes

         abstract 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.

        Parameters:
        subject - The NATS subject to publish to
        data - The raw byte data to publish
        headers - Optional headers to include

      • subscribeBytes

         abstract 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()
        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.
        Returns:

        a SubscriptionHandle; SubscriptionHandle.started is the readiness rendezvous

      • publishTyped

         abstract <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.

        Parameters:
        type - the type to resolve a codec from, which has to be T's

      • requestTyped

         abstract <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.

        Parameters:
        requestType - the type to resolve the request codec from, which has to be Req's
        responseType - the type to resolve the response codec from, which has to be Resp's

      • requestManyTyped

         abstract <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.

        Parameters:
        requestType - the type to resolve the request codec from, which has to be Req's
        responseType - the type to resolve the response codec from, which has to be Resp's

      • respondTyped

         abstract <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.

        Parameters:
        requestType - the type to resolve the request codec from, which has to be Req's
        responseType - the type to resolve the response codec from, which has to be Resp's