Docs navigation
Core
Build Plugins
AI
Dev & Test
Build
Data Access
Database Migration
Errors
Messaging
Analytics
API
Cloud
Configuration
Languages
Misc
Reactive
Views
Validation
Configuration Reference
NATS v5.2.0-SNAPSHOT
On this page

NATS

1 Introduction

This project includes integration between Micronaut and nats.io. The standard Java Client is used to do the actual publishing and consuming.

2 Release History

For this project, you can find a list of releases (with release notes) here:

https://github.com/micronaut-projects/micronaut-nats/releases

3 Using the Micronaut CLI

To create a project with NATS support using the Micronaut CLI, supply the nats feature to the features flag.

$ mn create-app my-nats-app --features nats

This will create a project with the minimum necessary configuration for NATS.

Messaging Application

The Micronaut CLI can generate messaging applications. This will create a Micronaut app with NATS support, and without an HTTP server (although you can add one if you desire). The profile also provides a couple commands for generating NATS consumers and producers.

To create a NATS messaging application, use the the following command:

$ mn create-messaging-app my-nats-service --features nats

As you’d expect, you can start the application with ./gradlew run (for Gradle) or ./mvnw compile exec:exec (Maven). The application will (with the default config) attempt to connect to NATS at nats://localhost:4222, and will continue to run without starting up an HTTP server. All communication to/from the service will take place via NATS producers and/or consumers.

Within the new project, you can now run the NATS specific code generation commands:

$ mn create-nats-producer Message
| Rendered template Producer.java to destination src/main/java/my/nats/app/MessageProducer.java

$ mn create-nats-listener Message
| Rendered template Listener.java to destination src/main/java/my/nats/app/MessageListener.java

4 NATS Quick Start

To add support for NATS.io to an existing project, you should first add the Micronaut NATS configuration to your build configuration. For example:

implementation("io.micronaut.nats:micronaut-nats")

Creating a NATS Producer with @NatsClient

To create a NATS Producer that sends messages you can simply define an interface that is annotated with @NatsClient.

For example the following is a trivial @NatsClient interface:

imports
import io.micronaut.nats.annotation.Subject;
import io.micronaut.nats.annotation.NatsClient;
@NatsClient // 
public interface ProductClient {

    @Subject("product") // 
    void send(byte[] data); // 
}

  1. The @NatsClient annotation is used to designate this interface as a client

  2. The @Subject annotation indicates which subject the Message should be published to

  3. It is also possible for the subject to be dynamic by making it a method argument

At compile time Micronaut will produce an implementation of the above interface. You can retrieve an instance of ProductClient either by looking up the bean from the ApplicationContext or by injecting the bean with @Inject:

productClient.send("quickstart".getBytes());

Creating a NATS Consumer with @NatsListener

To listen to NATS messages you can use the @NatsListener annotation to define a message listener.

The following example will listen for messages published by the ProductClient in the previous section:

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
@NatsListener // 
public class ProductListener {

    List<String> messageLengths = Collections.synchronizedList(new ArrayList<>());

    @Subject("product") // 
    public void receive(byte[] data) { // 
        messageLengths.add(new String(data));
        System.out.println("Java received " + data.length + " bytes from Nats");
    }
}

  1. The @NatsListener is used to designate this class as a listener.

  2. The @Subject annotation is again used to indicate which subject to subscribe to.

  3. The receive method defines one argument, which will receive the value.

5 Configuring the connection

All properties on the Options are available to be modified, either through configuration or a BeanCreatedEventListener.

The properties that can be converted from the string values in a configuration file can be configured directly.

Note
 Without any configuration the defaults in the Options will be used.
Tip
 It is also possible to disable the integration entirely with nats.enabled: false

Connections

It is possible to configure multiple connections to the same server, different servers, or a single connection to one of a list of servers.

nats.server1.addresses[0]=nats://localhost:4222
nats.server1.username=guest
nats.server1.password=guest
nats.server2.addresses[0]=nats://randomServer:4222
nats.server2.username=guest
nats.server2.password=guest

NATS also supports a fail over connection strategy where the first server that connects successfully will be used among a list of servers. To use this option in Micronaut, simply supply a list of host:port addresses.

nats.default.addresses[0]=nats://localhost:4222
nats.default.addresses[1]=nats://randomServer:4222
nats.default.username=guest
nats.default.password=guest
Important
 When the configuration option nats.servers is used, no other options underneath nats are read; for example nats.username.

If you need to setup TLS, it can be configured this way:

nats.default.addresses[0]=nats://localhost:4222
nats.default.tls.trust-store-path=/path/to/client.truststore.jks
nats.default.tls.trust-store-password=secret
nats.default.tls.certificate-path=/path/to/certificate.crt

  1. You can either use nats://localhost:4222 or tls://localhost:4222 as protocol.

  2. You can configure a complete truststore

  3. Or you can use a single certificate for connecting to NATS securely.

If the server requires a client certificate (mutual TLS), configure a key store holding the client certificate and its private key in addition:

nats.default.addresses[0]=tls://localhost:4222
nats.default.tls.trust-store-path=/path/to/client.truststore.jks
nats.default.tls.trust-store-password=secret
nats.default.tls.key-store-path=/path/to/client.p12
nats.default.tls.key-store-password=secret
nats.default.tls.key-store-type=PKCS12

  1. The key store with the client certificate and private key

  2. The password of the key store, also used for the private key

  3. The key store type, defaults to the JVM default key store type

Further connection options

Besides the options above, all options of the Options.Builder which can be converted from a configuration value can be set directly underneath the connection. Options without an argument, such as noRandomize(), are enabled by setting them to true.

nats.default.addresses[0]=nats://localhost:4222
nats.default.max-pings-out=5
nats.default.no-randomize=true
nats.default.request-cleanup-interval=10s
nats.default.reconnect-jitter=200ms
nats.default.max-messages-in-outgoing-queue=10000
nats.default.discard-messages-when-outgoing-queue-full=true
Note
 Options which require objects at runtime (for example executors, listeners or handlers) can only be set through a BeanCreatedEventListener for the io.micronaut.nats.connect.NatsConnectionFactoryConfig, using getBuilder().

6 NATS Producers

The example in the quick start presented a trivial definition of an interface that be implemented automatically for you using the @NatsClient annotation.

The implementation that powers @NatsClient (defined by the NatsIntroductionAdvice class) is, however, very flexible and offers a range of options for defining NATS clients.

6.1 Defining @NatsClient Methods

All methods that publish messages to NATS must meet the following conditions:

  • The method must reside in an interface annotated with @NatsClient.

  • The method or a method parameter must be annotated with @Subject.

  • The method must contain an argument representing the body of the message.

Important
 If a body argument cannot be found, an exception will be thrown.
Note
 In order for all of the functionality to work as designed in this guide your classes must be compiled with the parameters flag set to true. If your application was created with the Micronaut CLI, then that has already been configured for you.
Important
 Unless a reactive type is returned from the publishing method, the action is blocking.

6.1.1 Publishing Parameters

All options are available to be set for publishing messages. The publish method is used by the NatsIntroductionAdvice to publish messages and all arguments can be set through annotations or method arguments.

6.1.1.1 Subject

If you need to specify the subject of the message, apply the @Subject annotation to the method or an argument of the method. Apply the annotation to the method itself if the value is static for every execution. Apply the annotation to an argument of the method if the value should be set per execution.

imports
import io.micronaut.nats.annotation.NatsClient;
import io.micronaut.nats.annotation.Subject;
@NatsClient
public interface ProductClient {

    @Subject("product") // 
    void send(byte[] data);

    void send(@Subject String subject, byte[] data); // 
}

  1. The subject is static

  2. The subject must be set per execution

Producer Connection

If multiple Nats servers have been configured, the name of the server can be set in the @Subject annotation to designate which connection should be used to publish messages.

imports
import io.micronaut.nats.annotation.NatsClient;
import io.micronaut.nats.annotation.Subject;
@NatsClient
public interface ProductClient {

    @Subject(value = "product", connection = "product-cluster") // 
    void send(byte[] data);
}

  1. The connection is set on the subject annotation.

Note
 The connection option is also available to be set on the @NatsClient annotation.

Queues

Tip
 The NATS server will route the message to the queue and select a message receiver.

6.1.1.2 Headers

Headers can be set on the message with the @MessageHeader annotation applied to the method or an argument of the method. Apply the annotation to the method itself if the value is static for every execution. Apply the annotation to an argument of the method if the value should be set per execution.

imports
import io.micronaut.messaging.annotation.MessageBody;
import io.micronaut.messaging.annotation.MessageHeader;
import io.micronaut.nats.annotation.NatsClient;
import io.micronaut.nats.annotation.Subject;
import io.nats.client.impl.Headers;
@NatsClient
@MessageHeader(name = "x-product-sealed", value = "true") // 
@MessageHeader(name = "productSize", value = "large")
public interface ProductClient {

    @Subject("product")
    @MessageHeader(name = "x-product-count", value = "10") // 
    @MessageHeader(name = "productSize", value = "small")
    void send(byte[] data);

    @Subject("product")
    void send(@MessageHeader String productSize, // 
              @MessageHeader("x-product-count") Long count,
              byte[] data);

    @Subject("products")
    @MessageHeader(name = "x-product-count", value = "20")
    void send(@MessageBody byte[] data, @MessageHeader List<String> productSizes);// 

    @Subject("productHeader")
    void send(@MessageBody byte[] data, Headers headers);// 

}

  1. Headers can be defined at the class level and will apply to all methods. If a header is defined on the method with the same name as one on the class, the value on the method will be used.

  2. Multiple annotations can be used to set multiple headers on the method or class level.

  3. Headers can be set per execution. The name is inferred from the argument if the annotation value is not set. A null value results in the header not being set.

  4. You can also use a List as header. An empty list or a null value do not set the header.

  5. A Headers argument can be used to pass custom headers. Note: if the @MessageHeader is used on a method argument, the Headers argument will be ignored.

Note
 Python methods cannot be overloaded, so the Python client declares one method per signature (send, send_with_headers, send_with_product_sizes, send_with_nats_headers).

6.1.1.3 Message Body

Most examples up to this point have been using a byte[] as the body type for simplicity. This library supports most standard Java types and JSON serialization (using Jackson) by default. The functionality is extensible and it is possible to add support for additional types and serialization strategies. See the section on Message Serialization/Deserialization for more information.

7 NATS Consumers

The quick start section presented a trivial example of what is possible with the @NatsListener annotation.

The implementation that powers @NatsListener (defined by the NatsConsumerAdvice class) is, however, very flexible and offers a range of options for consuming NATS message.

7.1 Defining @NatsListener Methods

All methods that consume messages from NATS must meet the following conditions:

  • The method must reside in a class annotated with @NatsListener.

  • The method must be annotated with @Subject.

Note
 In order for all of the functionality to work as designed in this guide your classes must be compiled with the parameters flag set to true. If your application was created with the Micronaut CLI, then that has already been configured for you.

7.1.1 Consumer Parameters

The createDispatcher method is used by the NatsConsumerAdvice to consume messages. Some of the options can be directly configured through annotations.

Important
 In order for the consumer method to be invoked, all arguments must be satisfied. To allow execution of the method with a null value, the argument must be declared as nullable. If the arguments cannot be satisfied, the message will be rejected.

7.1.1.1 Subject

A @Subject annotation is required for a method to be a consumer of messages from Nats. Simply apply the annotation to the method and supply the name of the subject you would like to listen to.

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
@NatsListener
public class ProductListener {

    List<Integer> messageLengths = Collections.synchronizedList(new ArrayList<>());

    @Subject("product") // 
    public void receive(byte[] data) {
        messageLengths.add(data.length);
    }
}

  1. The subject annotation is set per method. Multiple methods may be defined with different subjects in the same class.

Queue Support

Subscribers may specify queue groups at subscription time. When a message is published to the group, NATS will deliver it to a one-and-only-one subscriber.

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
@NatsListener
public class ProductListener {

    List<String> messageLengths = Collections.synchronizedList(new ArrayList<>());

    @Subject(value = "product", queue = "product-queue") // 
    public void receiveByQueue1(byte[] data) {
        messageLengths.add(new String(data));
        System.out.println("Java received " + data.length + " bytes from Nats");
    }

    @Subject(value = "product", queue = "product-queue")
    public void receiveByQueue2(byte[] data) {
        messageLengths.add(new String(data));
        System.out.println("Java received " + data.length + " bytes from Nats");
    }
}

  1. Defining a queue can be done in the @Subject

Important
 Queue groups do not persist messages. If no listeners are available, the message is discarded.

Other Options

If multiple Nats servers have been configured, the name of the server can be set in the @Subject annotation to designate which connection should be used to listen for messages.

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
@NatsListener
public class ProductListener {

    List<String> messageLengths = Collections.synchronizedList(new ArrayList<>());

    @Subject(value = "product", connection = "product-cluster") // 
    public void receive(byte[] data) {
        messageLengths.add(new String(data));
        System.out.println("Java received " + data.length + " bytes from Nats");
    }
}

  1. The connection is set on the subject annotation.

Note
 The connection option is also available to be set on the @NatsListener annotation.

7.1.1.2 Headers

Headers can be retrieved with the @MessageHeader annotation applied to the arguments of the method.

imports
import io.micronaut.messaging.annotation.MessageBody;
import io.micronaut.messaging.annotation.MessageHeader;
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
import io.nats.client.impl.Headers;
@NatsListener
public class ProductListener {

    Set<String> messageProperties = Collections.synchronizedSet(new HashSet<>());

    @Subject("product")
    public void receive(byte[] data,
            @MessageHeader("x-product-sealed") Boolean sealed, // 
            @MessageHeader("x-product-count") Long count, // 
            @Nullable @MessageHeader String productSize) { // 
        messageProperties.add(sealed + "|" + count + "|" + productSize);
    }

    @Subject("products")
    public void receive(@MessageBody byte[] data, @MessageHeader("x-product-sealed") Boolean sealed,
            @MessageHeader("x-product-count") Long count, @MessageHeader List<String> productSizes) { // 
        for (String productSize : productSizes) {
            messageProperties.add(sealed + "|" + count + "|" + productSize);
        }
    }

    @Subject("productHeader")
    public void receive(@MessageBody byte[] data, Headers headers) { // 
        String productSize = headers.get("productSize").get(0);
        messageProperties.add(
                headers.get("x-product-sealed").get(0) + "|" +
                        headers.get("x-product-count").get(0) + "|" +
                        productSize);
    }
}

  1. The header name comes from the annotation and the value is retrieved and converted to a Boolean.

  2. The header name comes from the annotation and the value is retrieved and converted to a Long.

  3. The header name comes from the argument name. This argument allows null values.

  4. The header can also be a list representing multiple values.

  5. All headers can be bound to a Headers argument.

Note
 Python methods cannot be overloaded, so the Python listener declares one method per subject (receive, receive_product_sizes, receive_headers).

7.1.1.3 Nats Types

Arguments can also be bound based on their type. Several types are supported by default and each type has a corresponding NatsTypeArgumentBinder. The argument binders are covered in detail in the section on Custom Parameter Binding.

There is only type that is supported for retrieving data about the Message.

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
import io.nats.client.Connection;
import io.nats.client.Message;
import io.nats.client.Subscription;
import io.nats.client.impl.Headers;
@NatsListener
public class ProductListener {

    List<String> messages = Collections.synchronizedList(new ArrayList<>());

    @Subject("product")
    public void receive(byte[] data,
            Message message,
            Connection connection,
            Subscription subscription,
            Headers headers) { // 
        messages.add(String.format("subject: [%s], maxPayload: [%s], subscriptionSubject: [%s], x-productCount: [%s]",
                message.getSubject(),
                connection.getMaxPayload(), subscription.getSubject(),
                headers.get("x-product-count").get(0)));
    }
}

  1. The arguments are bound from the Message.

7.1.1.4 Message Body

Most examples up to this point have been using a byte[] as the body type for simplicity. This library supports most standard Java types and JSON deserialization (using Jackson) by default. The functionality is extensible and it is possible to add support for additional types and deserialization strategies. See the section on Message Serialization/Deserialization for more information.

7.1.1.5 Custom Parameter Binding

Default Binding Functionality

Consumer argument binding is achieved through an ArgumentBinderRegistry that is specific for binding consumers from Nats messages. The class responsible for this is the NatsBinderRegistry.

The registry supports argument binders that are used based on an annotation applied to the argument or the argument type. All argument binders must implement either NatsAnnotatedArgumentBinder or NatsTypeArgumentBinder. The exception to that rule is the NatsDefaultBinder which is used when no other binders support a given argument.

When an argument needs bound, the Message is used as the source of all of the available data. The binder registry follows a small sequence of steps to attempt to find a binder that supports the argument.

  1. Search the annotation based binders for one that matches any annotation on the argument that is annotated with @Bindable.

  2. Search the type based binders for one that matches or is a subclass of the argument type.

  3. Return the default binder.

The default binder binds the body of the message to the argument.

Custom Binding

To inject your own argument binding behavior, it is as simple as registering a bean. The existing binder registry will inject it and include it in the normal processing.

Annotation Binding

A custom annotation can be created to bind consumer arguments. A custom binder can then be created to use that annotation and the Message to supply a value for the argument. The value may in fact come from anywhere, however for the purposes of this documentation, the replyTo in the message is used.

imports
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

import io.micronaut.core.bind.annotation.Bindable;
@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.PARAMETER})
@Bindable // 
public @interface SID {
}

  1. The @Bindable annotation is required for the annotation to be considered for binding.

imports
import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.nats.bind.NatsAnnotatedArgumentBinder;
import io.nats.client.Message;
import jakarta.inject.Singleton;
@Singleton // 
public class SIDAnnotationBinder implements NatsAnnotatedArgumentBinder<SID> { // 

    private final ConversionService conversionService;

    public SIDAnnotationBinder(ConversionService conversionService) { // 
        this.conversionService = conversionService;
    }

    @Override
    public Class<SID> getAnnotationType() {
        return SID.class;
    }

    @Override
    public BindingResult<Object> bind(ArgumentConversionContext<Object> context, Message source) {
        String sid = source.getSID(); // 
        return () -> conversionService.convert(sid, context); // 
    }
}

  1. The class is made a bean by annotating with @Singleton.

  2. The custom annotation is used as the generic type for the interface.

  3. The conversion service is injected into the instance.

  4. The replyTo is retrieved from the message state.

  5. The replyTo is converted to the argument type.

The annotation can now be used on the argument in a consumer method.

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
@NatsListener
public class ProductListener {

    List<String> messages = Collections.synchronizedList(new ArrayList<>());

    @Subject("product")
    public void receive(byte[] data, @SID String sid) { // 
        messages.add(sid);
    }
}

Type Binding

A custom binder can be created to support any argument type. For example the following class could be created to bind values from the headers. This functionality could allow the work of retrieving and converting the headers to occur in a single place instead of multiple times in your code.

imports
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;
public class ProductInfo {

    private String size;
    private Long count;
    private Boolean sealed;

    public ProductInfo(@Nullable String size, // 
                       @NonNull Long count, // 
                       @NonNull Boolean sealed) { // 
        this.size = size;
        this.count = count;
        this.sealed = sealed;
    }

    public String getSize() {
        return size;
    }

    public Long getCount() {
        return count;
    }

    public Boolean getSealed() {
        return sealed;
    }
}

  1. The size argument is not required

  2. The count argument is required

  3. The sealed argument is required

A type argument binder can then be created to create the ProductInfo instance to bind to your consumer method argument.

imports
import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.convert.ConversionError;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.core.type.Argument;
import io.micronaut.nats.bind.NatsHeaderConvertibleValues;
import io.micronaut.nats.bind.NatsTypeArgumentBinder;
import io.nats.client.Message;
import io.nats.client.impl.Headers;
import jakarta.inject.Singleton;
@Singleton // 
public class ProductInfoTypeBinder implements NatsTypeArgumentBinder<ProductInfo> { //

    private final ConversionService conversionService;

    ProductInfoTypeBinder(ConversionService conversionService) { //
        this.conversionService = conversionService;
    }

    @Override
    public Argument<ProductInfo> argumentType() {
        return Argument.of(ProductInfo.class);
    }

    @Override
    public BindingResult<ProductInfo> bind(ArgumentConversionContext<ProductInfo> context, Message source) {
        Headers rawHeaders = source.getHeaders(); //

        if (rawHeaders == null) {
            return BindingResult.EMPTY;
        }

        NatsHeaderConvertibleValues headers = new NatsHeaderConvertibleValues(rawHeaders, conversionService);

        String size = headers.get("productSize", String.class).orElse(null);  //
        Optional<Long> count = headers.get("x-product-count", Long.class); //
        Optional<Boolean> sealed = headers.get("x-product-sealed", Boolean.class); // 

        if (headers.getConversionErrors().isEmpty() && count.isPresent() && sealed.isPresent()) {
            return () -> Optional.of(new ProductInfo(size, count.get(), sealed.get())); //
        } else {
            return new BindingResult<ProductInfo>() {
                @Override
                public Optional<ProductInfo> getValue() {
                    return Optional.empty();
                }

                @Override
                public List<ConversionError> getConversionErrors() {
                    return headers.getConversionErrors(); //
                }
            };
        }
    }
}

  1. The class is made a bean by annotating with @Singleton.

  2. The custom type is used as the generic type for the interface.

  3. The conversion service is injected into the instance.

  4. The headers are retrieved from the message state.

  5. The productSize header is retrieved, defaulting to null if the value was not found or could not be converted.

  6. The x-product-count header is retrieved and converted with a new argument context that is used to retrieve conversion errors later.

  7. The x-product-sealed header is retrieved and converted with a new argument context that is used to retrieve conversion errors later.

  8. There are no conversion errors and the two required arguments are present, so the instance can be constructed.

  9. There are conversion errors or one of the required arguments is not present so a custom BindingResult is returned that allows the conversion errors to be handled appropriately.

8 Request-Reply (RPC)

This library supports RPC through the usage of Request-Reply. Both blocking and non blocking variations are supported.

The following is an example direct reply to where the consumer is converting the body to upper case and replying with the converted string.

Client Side

The "client side" in this case starts by publishing a message. A consumer somewhere will then receive the message and reply with a new value.

imports
import io.micronaut.nats.annotation.NatsClient;
import io.micronaut.nats.annotation.Subject;
import org.reactivestreams.Publisher;
@NatsClient
public interface ProductClient {

    @Subject("product")
    String send(String data); // 

    @Subject("product")
    Publisher<String> sendReactive(String data); // 
}

  1. The send method is blocking and will return when the response is received.

  2. The sendReactive method returns a reactive type that will complete when the response is received. Reactive methods will be executed on the IO thread pool.

Important
 In order for the publisher to assume RPC should be used instead of just completing when the publish is confirmed, the data type must not be Void. In both cases above, the data type is String.

Server Side

The "server side" in this case starts with the consumption of a message, and then a new message is published by returning the result

imports
import io.micronaut.nats.annotation.NatsListener;
import io.micronaut.nats.annotation.Subject;
@NatsListener
public class ProductListener {

    @Subject("product")
    public String toUpperCase(String data) { // 
        return data.toUpperCase(); // 
    }
}

  1. The data from message is injected.

  2. The converted message is returned.

Note
 If the reply publish fails for any reason, the original message will be rejected.
Important
 RPC consumer methods must never return a reactive type. Because the resulting publish needs to occur on the same thread and only a single item can be emitted, there is no value in doing so.

9 Message Serialization/Deserialization (SerDes)

The serialization and deserialization of message bodies is handled through instances of NatsMessageSerDes. The ser-des (Serializer/Deserializer) is responsible for both serialization and deserialization of Nats message bodies into the message body types defined in your clients and consumers methods.

The ser-des are managed by a NatsMessageSerDesRegistry. All ser-des beans are injected in order into the registry and then searched for when serialization or deserialization is needed. The first ser-des that returns true for supports-java.lang.Class- is returned and used.

By default, standard Java lang types and JSON format (with Jackson) are supported. You can supply your own ser-des by simply registering a bean of type NatsMessageSerDes. All ser-des implement the Ordered interface, so custom implementations can come before, after, or in between the default implementations.

9.1 Custom SerDes

A custom serializer/deserializer would be necessary to support custom data formats. In the section on Custom Consumer Binding an example was demonstrated that allowed binding a ProductInfo type from the headers of the message. If instead that object should represent the body of the message with a custom data format, you could register your own serializer/deserializer to do so.

In this example a simple data format of the string representation of the fields are concatenated together with a pipe character.

imports
import io.micronaut.core.convert.ConversionService;
import io.micronaut.core.type.Argument;
import io.micronaut.nats.serdes.NatsMessageSerDes;
import io.nats.client.Message;
import jakarta.inject.Singleton;

import java.nio.charset.StandardCharsets;
import java.util.Optional;
@Singleton // 
public class ProductInfoSerDes implements NatsMessageSerDes<ProductInfo> { // 

    private final ConversionService conversionService;

    public ProductInfoSerDes(ConversionService conversionService) { // 
        this.conversionService = conversionService;
    }

    @Override
    public ProductInfo deserialize(Message message, Argument<ProductInfo> argument) { // 
        String body = new String(message.getData(), StandardCharsets.UTF_8);
        String[] parts = body.split("\\|");
        if (parts.length == 3) {
            String size = parts[0];
            if (size.equals("null")) {
                size = null;
            }

            Optional<Long> count = conversionService.convert(parts[1], Long.class);
            Optional<Boolean> sealed = conversionService.convert(parts[2], Boolean.class);

            if (count.isPresent() && sealed.isPresent()) {
                return new ProductInfo(size, count.get(), sealed.get());
            }
        }
        return null;
    }

    @Override
    public byte[] serialize(ProductInfo data) { // 
        if (data == null) {
            return null;
        }
        return (data.getSize() + "|" + data.getCount() + "|" + data.getSealed()).getBytes(StandardCharsets.UTF_8);
    }

    @Override
    public boolean supports(Argument<ProductInfo> argument) { // 
        return argument.getType().isAssignableFrom(ProductInfo.class);
    }
}

  1. The class is declared as a singleton so it will be registered with the context

  2. The generic specifies what type we want to accept and return

  3. The conversion service is injected to convert the parts of the message to the required types

  4. The deserialize method takes the bytes from the message and constructs a ProductInfo

  5. The serialize method takes the ProductInfo and returns the bytes to publish. A mutable version of the properties is also provided so properties such as the content type can be set before publishing.

  6. The supports method ensures only the correct body types are processed by this ser-des

Tip
 Because the getOrder method was not overridden, the default order of 0 is used. All default ser-des have a lower precedent than the default order which means this ser-des will be checked before the others.

10 NATS Health Indicator

This library comes with a health indicator for applications that are using the management module in Micronaut. See the Health Endpoint documentation for more information about the endpoint itself.

The information reported from the health indicator is under the nats key.

"nats": {
  "status": "UP",
  "details": {
    "servers": ["nats://localhost:4222"]
  }
}
Tip
 To disable the NATS health indicator entirely, add endpoints.health.nats.enabled: false.

11 Jetstream

Jetstream is built-in distributed persistence system built into Nats.io which enables new functionalities like

  • fault-tolerance

  • replication

  • exactly-once semantics

  • replay policies

  • retention policy and limits

  • streaming

11.1 Streams

Streams are 'message stores', each stream defines how messages are stored and what the limits (duration, size, interest) of the retention are. Streams consume normal NATS subjects, any message published on those subjects will be captured in the defined storage system. You can do a normal publish to the subject for unacknowledged delivery, though it’s better to use the JetStream publish calls instead as the JetStream server will reply with an acknowledgement that it was successfully stored.

For further information have a look at JetStream Model Deep Dive.

11.1.1 Configuration

All properties on the JetStreamOptions.Builder and StreamConfiguration.Builder are available to be modified, either through configuration or a BeanCreatedEventListener.

The following properties are available for a stream configuration:

The options of the JetStreamOptions.Builder are configured underneath jetstream:

A simple configuration for jetstream and a single stream can look like:

nats.default.jetstream.streams.events.storage-type=Memory
nats.default.jetstream.streams.events.subjects[0]=events.>

Subject transform, compression and metadata

nats.default.jetstream.streams.events.subjects[0]=events.>
nats.default.jetstream.streams.events.compression-option=s2
nats.default.jetstream.streams.events.metadata.owner=team-a
nats.default.jetstream.streams.events.subject-transform.source=events.*.*
nats.default.jetstream.streams.events.subject-transform.destination=events.{{wildcard(2)}}.{{wildcard(1)}}

  1. Compression of the stream storage, either none or s2

  2. Custom metadata of the stream

  3. Transformation of the subjects of incoming messages

Placement, republish and consumer limits

nats.default.jetstream.streams.events.subjects[0]=events.>
nats.default.jetstream.streams.events.placement.cluster=east
nats.default.jetstream.streams.events.placement.tags[0]=ssd
nats.default.jetstream.streams.events.republish.source=events.>
nats.default.jetstream.streams.events.republish.destination=republished.>
nats.default.jetstream.streams.events.republish.headers-only=false
nats.default.jetstream.streams.events.consumer-limits.inactive-threshold=10m
nats.default.jetstream.streams.events.consumer-limits.max-ack-pending=1000

  1. The cluster and tags the stream should be placed on

  2. Republishes stored messages to another subject

  3. Default limits for consumers of the stream

Mirror

A stream can mirror another stream. A mirror can not have subjects of its own.

nats.default.jetstream.streams.events-mirror.mirror.name=events
nats.default.jetstream.streams.events-mirror.mirror.filter-subject=events.orders.>
nats.default.jetstream.streams.events-mirror.mirror.start-seq=1
nats.default.jetstream.streams.events-mirror.mirror.domain=hub
nats.default.jetstream.streams.events-mirror.mirror.subject-transforms[0].source=events.orders.>
nats.default.jetstream.streams.events-mirror.mirror.subject-transforms[0].destination=mirror.orders.>

  1. The name of the stream to mirror

  2. Only mirror a subset of the messages. Can not be combined with subject-transforms

  3. The JetStream domain of the origin stream. Alternatively external.api and external.deliver can be configured

  4. Transformations of the subjects while mirroring, the sources act as filter

Sources

A stream can be sourced from one or more other streams, in addition to its own subjects.

nats.default.jetstream.streams.all-events.sources[0].name=events
nats.default.jetstream.streams.all-events.sources[0].subject-transforms[0].source=events.*
nats.default.jetstream.streams.all-events.sources[0].subject-transforms[0].destination=all.events.$1
nats.default.jetstream.streams.all-events.sources[1].name=remote-events
nats.default.jetstream.streams.all-events.sources[1].filter-subject=remote.>
nats.default.jetstream.streams.all-events.sources[1].external.api=$JS.remote.API
nats.default.jetstream.streams.all-events.sources[1].external.deliver=deliver.remote

  1. The name of the stream to source from

  2. Transformations of the subjects while sourcing, the sources act as filter. Can not be combined with filter-subject

  3. Source a stream of another account or domain

11.1.2 Producer

The example in the quick start presented a trivial definition of an interface that be implemented automatically for you using the @JetstreamClient annotation.

The implementation that powers @JetstreamClient (defined by the JetStreamIntroductionAdvice class) is, however, very flexible and offers a range of options for defining Jetstream clients.

The @JetstreamClient extends the default @NatsClient and is based on the same methods. So you can still use all header and subject functionalities as you already know.

@JetstreamClient have a special extension for the options you want to publish.

imports
import io.micronaut.messaging.annotation.MessageBody;
import io.micronaut.nats.annotation.Subject;
import io.micronaut.nats.jetstream.annotation.JetStreamClient;
import io.nats.client.PublishOptions;
import io.nats.client.api.PublishAck;
@JetStreamClient
public interface ProductClient {

    PublishAck send(@Subject String subject, @MessageBody byte[] data, PublishOptions publishOptions); // 
}

  1. With PublishOptions you can define additional options for publishing message with jetstream. You can also retrieve a PublishAck object representing a JetStream enabled server acknowledgment.

11.1.3 Consumer

A consumer is a stateful view of a stream. It acts as interface for clients to consume a subset of messages stored in a stream and will keep track of which messages were delivered and acknowledged by clients.

Unlike with core NATS which provides an at most once delivery guarantee of a message, a consumer can provide an at least once delivery guarantee. This is achieved by the combination of published messages being persisted to the stream as well as the consumer tracking delivery and acknowledgement of each individual message as clients receive and process them. JetStream consumers support multiple kinds of acknowledgements and multiple acknowledgement policies. They will take care of automatically re-deliver un-acked (or 'nacked') messages up to a user specified maximum number of delivery attempts (there is an advisory being emitted when a message reaches this limit).

Consumers can be push-based where messages will be delivered to a specified subject or pull-based which allows clients to request batches of messages on demand. The choice of what kind of consumer to use depends on the use-case but typically in the case of a client application that needs to get their own individual replay of messages from a stream you would use an 'ordered push consumer'. If there is a need to process messages and easily scale horizontally, you would use a 'pull consumer'.

In addition to the choice of being push or pull, a consumer can also be ephemeral or durable. A consumer is considered durable when an explicit name is set on the Durable field when creating the consumer, otherwise it is considered ephemeral. Durables and ephemeral behave exactly the same except that an ephemeral will be automatically cleaned up (deleted) after a period of inactivity, specifically when there are no subscriptions bound to the consumer. By default, durables will remain even when there are periods of inactivity (unless InactiveThreshold is set explicitly).

11.1.3.1 Push based

A push consumer is where the server is in control and sends messages to the client. It can be made durable or ephemeral based on your use case.

Push consumers are very similiar to the already known @NatsListener. Let’s look at a quick example.

imports
import io.micronaut.nats.jetstream.annotation.JetStreamListener;
import io.micronaut.nats.jetstream.annotation.PushConsumer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@JetStreamListener // 
public class ProductListener {

    List<byte[]> messageLengths = Collections.synchronizedList(new ArrayList<>());

    @PushConsumer(value = "myevents", subject = "myevents.>", durable = "test") // 
    public void receive(byte[] data) {
        messageLengths.add(data);
    }
}

  1. The class needs to be annotated with @JetstreamListener.

  2. The @PushConsumer configures the stream(value) and the subject to listen for. It is possible to use all known values from SubscribeOptions and PushSubscribeOptions such as durable.

Tip
 To consume multiple subjects of a stream with a single consumer, use filterSubjects, for example @PushConsumer(value = "events", durable = "orders-and-payments", filterSubjects = {"events.orders.>", "events.payments.>"}). With filterSubjects the subject is optional; if it is set, it has to be one of the filter subjects. The single valued filterSubject is deprecated.

11.1.3.2 Pull based

A pull consumer allows you to control when the server sends the client messages.

For creating a new pull subscription you need to inject PullConsumerRegistry.

Let’s look at a quick example.

imports
import io.micronaut.nats.jetstream.PullConsumerRegistry;
import io.nats.client.JetStreamApiException;
import io.nats.client.JetStreamSubscription;
import io.nats.client.Message;
import io.nats.client.PullSubscribeOptions;
import io.nats.client.api.ConsumerConfiguration;
@Singleton
public class PullConsumerHelper {

    private final PullConsumerRegistry pullConsumerRegistry;

    public PullConsumerHelper(PullConsumerRegistry pullConsumerRegistry) { // 
        this.pullConsumerRegistry = pullConsumerRegistry;
    }

    public List<Message> pullMessages() throws JetStreamApiException, IOException {
        PullSubscribeOptions pullSubscribeOptions =
            PullSubscribeOptions.builder()
                                .stream("events")
                                .configuration(
                                    ConsumerConfiguration
                                        .builder()
                                        .ackWait(
                                            Duration.ofMillis(
                                                2500))
                                        .build())
                                .build();
        JetStreamSubscription jetStreamSubscription =
            pullConsumerRegistry.newPullConsumer("events.>", pullSubscribeOptions); // 

        List<Message> messages = jetStreamSubscription.fetch(2, Duration.ofSeconds(2L)); // 
        messages.forEach(Message::ack); // 
        return messages;
    }
}

  1. For creating a new pull subscription you need to inject PullConsumerRegistry.

  2. Create a new JetStreamSubscription with the subjects you want to subscribe to and the necessary PullSubscribeOptions.

  3. Fetch the messages from the subscription.

  4. Acknowledge the fetched messages.

11.2 Key/Value Store

JetSteam, the persistence layer of NATS, doesn’t just allow for higher qualities of service and features associated with 'streaming', but it also enables some functionalities not found in messaging systems.

One such feature is the Key/Value store functionality, which allows client applications to create 'buckets' and use them as immediately consistent, persistent associative arrays.

You can use KV buckets to perform the typical operations you would expect from an immediately consistent key/value store:

  • put: associate a value with a key

  • get: retrieve the value associated with a key

  • delete: clear any value associated with a key

  • purge: clear all the values associated with all keys

  • create: associate the value with a key only if there is currently no value associated with that key (i.e. compare to null and set)

  • update: compare and set (aka compare and swap) the value for a key

  • keys: get a copy of all the keys (with a value or operation associated to it)

You can set limits for your buckets, such as: - the maximum size of the bucket - the maximum size for any single value - a TTL: how long the store will keep values for

Finally, you can even do things that typically can not be done with a Key/Value Store:

  • watch: watch for changes happening for a key, which is similar to subscribing (in the publish/subscribe sense) to the key: the watcher receives updates due to put or delete operations on the key pushed to it in real-time as they happen

  • watch all: watch for all the changes happening on all the keys in the bucket

  • history: retrieve a history of the values (and delete operations) associated with each key over time (by default the history of buckets is set to 1, meaning that only the latest value/operation is stored)

11.2.1 Configuration

All properties on the JetStreamOptions.Builder and KeyValueConfiguration.Builder are available to be modified, either through configuration or a BeanCreatedEventListener.

The following properties are available for a key value configuration:

A simple configuration for key value can look like:

nats.default.jetstream.keyvalue.examplebucket.storage-type=Memory
nats.default.jetstream.keyvalue.examplebucket.max-history-per-key=5

Compression, metadata, placement and republish can be configured as for streams:

nats.default.jetstream.keyvalue.examplebucket.compression=true
nats.default.jetstream.keyvalue.examplebucket.metadata.owner=team-a
nats.default.jetstream.keyvalue.examplebucket.placement.cluster=east
nats.default.jetstream.keyvalue.examplebucket.republish.source=$KV.examplebucket.>
nats.default.jetstream.keyvalue.examplebucket.republish.destination=changes.examplebucket.>

A bucket can also mirror another bucket, or be sourced from other buckets. The names refer to the buckets, the KV_ prefix of the underlying streams is added automatically.

nats.default.jetstream.keyvalue.examplebucket-mirror.mirror.name=examplebucket
nats.default.jetstream.keyvalue.aggregated-bucket.sources[0].name=examplebucket
nats.default.jetstream.keyvalue.aggregated-bucket.sources[1].name=otherbucket
Note
 The entries are mirrored into the stream KV_<mirror bucket> keeping the subjects of the origin bucket, so reading them through the mirror bucket name does not return them. Read through the origin bucket instead; the mirror is allowed to answer those reads (mirror direct).

11.2.2 Usage

Nats.io provides an KeyValue interface for the usage of Key/Value Stores.

To use it, just inject your Key/Value Store as follows:

imports
import io.micronaut.nats.jetstream.annotation.KeyValueStore;
import io.nats.client.JetStreamApiException;
import io.nats.client.KeyValue;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;

import java.io.IOException;
@Singleton
public class KeyValueStoreHolder {

    @Inject
    @KeyValueStore("examplebucket")
    KeyValue store; // 

    public void put(String key, String value) throws JetStreamApiException, IOException {
        store.put(key, value);
    }

}

  1. Simply inject the key/value store via the @KeyValueStore annotation.

11.3 Object Store (Experimental)

Warning
 Experimental Preview

The Object Store allows you to store data of any (i.e. large) size by implementing a chunking mechanism, allowing you to for example store and retrieve files (i.e. the object) of any size by associating them with a path and a file name (i.e. the key). You obtain a ObjectStoreManager object from your JetStream context.

11.3.1 Configuration

All properties on the JetStreamOptions.Builder and ObjectStoreConfiguration.Builder are available to be modified, either through configuration or a BeanCreatedEventListener.

A simple configuration for an object store can look like:

nats.default.jetstream.objectstore.examplebucket.storage-type=Memory

Compression, metadata and placement can be configured as well:

nats.default.jetstream.objectstore.examplebucket.storage-type=Memory
nats.default.jetstream.objectstore.examplebucket.compression=true
nats.default.jetstream.objectstore.examplebucket.metadata.owner=team-a
nats.default.jetstream.objectstore.examplebucket.placement.cluster=east
nats.default.jetstream.objectstore.examplebucket.placement.tags[0]=ssd

11.3.2 Usage

Nats.io provides an ObjectStore interface for the usage of Object Stores.

To use it, just inject your Object Store as follows:

imports
import io.micronaut.nats.jetstream.annotation.ObjectStore;
import io.nats.client.JetStreamApiException;
import jakarta.inject.Inject;
import jakarta.inject.Singleton;

import java.io.IOException;
import java.io.InputStream;
import java.security.NoSuchAlgorithmException;
@Singleton
public class ObjectStoreHolder {

    @Inject
    @ObjectStore("examplebucket")
    io.nats.client.ObjectStore store; // 

    public void put(String key, InputStream inputStream) throws JetStreamApiException, IOException, NoSuchAlgorithmException {
        store.put(key, inputStream);
    }

}

  1. Simply inject the object store via the @ObjectStore annotation.

12 Repository

You can find the source code of this project in this repository:

https://github.com/micronaut-projects/micronaut-nats