NATS Spring Support

Current Release: 0.6.2+3.5 Current Snapshot: 0.6.3+3.5-SNAPSHOT

Please note the version number is a combination of Semver and the Spring Boot Version, <major.minor.patch>-<sbMajor.sbMinor>

Using the Starter
Using the NATS Modules
- Multiple NATS Connections
Using the Binder
Configuration
- Custom Listeners
Samples
Building This Project
External Resources
License

This repository contains two core packages:

nats-spring an implementation of the autoconfigure pattern for a NATS connection using the core NATS Java Client
nats-spring-cloud-stream-binder a spring cloud binder for NATS

A third package nats-spring-samples is included to demonstrate how the other two modules can be used.

Version Notes

As of version 0.3.x the properties used to configure can be in YAML or properties if you pass in the NATS properties externally. If you want the application file to contain connection info it should be a properties file and not YAML. In other words:

spring.cloud.stream.bindings.input.destination=dataIn
spring.cloud.stream.bindings.input.binder=nats1
spring.cloud.stream.binders.nats1.type=nats
spring.cloud.stream.binders.nats1.environment.nats.spring.cloud.stream.binder.server=nats://localhost:4222

it works while the YAML equivalent will not.

Snapshots are hosted on central.sonatype.org, to access these within maven update your settings to include:

<profiles>
    <profile>
        <id>allow-snapshots</id>
        <activation>
            <activeByDefault>true</activeByDefault>
        </activation>
        <repositories>
            <repository>
                <id>sonatype-snapshots-repo</id>
                <url>https://central.sonatype.com/repository/maven-snapshots</url>
                <releases>
                    <enabled>false</enabled>
                </releases>
                <snapshots>
                    <enabled>true</enabled>
                </snapshots>
            </repository>
        </repositories>
    </profile>
</profiles>

The released version should be available at maven central. If the release has not propagated yet, it can be found at

<repository>
    <id>central-sonatype</id>
    <name>central-sonatype</name>
    <url>https://repo1.maven.org/maven2/</url>
</repository>

and ultimately at maven central.

Using the Starter

A spring boot starter is provided that will bring in the autoconfigure module.

To include the starter, add the following dependency to your pom.xml:

<dependency>
    <groupId>io.nats</groupId>
    <artifactId>nats-spring-boot-starter</artifactId>
    <version>major.minor.patch[-SNAPSHOT]</version>
</dependency>

Using the NATS Modules

To depend on the autoconfigure module, add it as a dependency in your pom.xml:

<dependency>
    <groupId>io.nats</groupId>
    <artifactId>jnats</artifactId>
    <version>2.21.1</version>
</dependency>
<dependency>
    <groupId>io.nats</groupId>
    <artifactId>nats-spring</artifactId>
    <version>major.minor.patch[-SNAPSHOT]</version>
</dependency>

This module defines and exports an autoconfiguration for NATS Connection objects.

Multiple NATS Connections

By default, the autoconfigure modules will create a single NATS connection and the binders will create a second one.

An example multi-binder-sample demonstrates how to have multiple named binders.

Using the NATS Cloud Stream Binder

To use the binder, define a binding in your application.yml with the type nats:

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: dataIn
          binder: nats
        output:
          destination: dataOut
          binder: nats

and include a dependency on the library:

<dependency>
    <groupId>io.nats</groupId>
    <artifactId>nats-spring-cloud-stream-binder</artifactId>
    <version>major.minor.patch[-SNAPSHOT]</version>
</dependency>

See the Spring Cloud Stream Documentation for information about more complex configurations of the bindings.

Spring Cloud Stream functional applications are supported. Bind Consumer, Function, or StreamBridge applications with the standard functional binding names such as input-in-0, transform-in-0, and transform-out-0; see nats-spring-samples and demo for working examples.

The NATS binder leverages the autoconfigure module, or manual configuration to build a NATS connection. Standard properties are used to specify inputs and outputs. Inputs, specified with a destination and group name are mapped to subjects and queue names, with the destination becoming the subject, and the group becoming the queue. Outputs are specified with a destination name that becomes the subject.

Core NATS event consumers are implemented with a dispatcher, resulting in a thread per consumer. JetStream event consumers use a named JetStream consumer context. Polled consumers use a subscription for core NATS and a named JetStream consumer context for JetStream.

Producers publish directly through the connection.

JetStream

JetStream is opt-in per binding. Producer stream provisioning is optional with provision-stream=true; consumer streams and consumers should be created with NATS or the jnats ConsumerConfiguration API before the binder starts. JetStream consumers require stream-name and either consumer-name or a binding group. The binding group is used as the JetStream consumer name when consumer-name is not set.

nats.spring.cloud.stream.bindings.output.producer.jet-stream=true
nats.spring.cloud.stream.bindings.output.producer.stream-name=ORDERS
nats.spring.cloud.stream.bindings.output.producer.provision-stream=true
nats.spring.cloud.stream.bindings.output.producer.stream-storage-type=memory
nats.spring.cloud.stream.bindings.output.producer.stream-replicas=1
nats.spring.cloud.stream.bindings.input.consumer.jet-stream=true
nats.spring.cloud.stream.bindings.input.consumer.stream-name=ORDERS
nats.spring.cloud.stream.bindings.input.consumer.consumer-name=orders-worker

When producer provisioning is enabled, the binder creates a missing stream for the binding destination subject. If the stream already exists, it must already cover the destination subject directly or through a NATS wildcard subject. stream-storage-type and stream-replicas are used only when creating a missing stream.

JetStream producers wait for the server publish acknowledgement. Event consumers acknowledge after the Spring output channel accepts the message and negatively acknowledge when the output channel rejects the message. Polled consumers use the jnats consumer context and acknowledge through Spring’s poll acknowledgement callback: successful polls acknowledge and rejected polls negatively acknowledge for redelivery.

Consumer delivery settings such as acknowledgement wait, maximum deliveries, flow control, replay policy, and pull limits are configured on the JetStream consumer itself. Applications that previously configured those settings as binder consumer properties should create or update the JetStream consumer before the binding starts and set consumer-name to that consumer name.

JetStream publishing does not support Spring reply channels because NATS uses the reply subject for JetStream publish acknowledgements. Use core NATS bindings for request-reply.

Request-Reply

The binder uses message header propagation to support NATS-style request-reply. The reply-to subject is passed as a header when a message is received. That reply to header is used as the outgoing subject if it is present when a message is sent. This system by-passes the configured producer destination.

Partitions

NATS does not use the concept of partitions. The binder supports the API for partitions as far as required, but no underlying messaging is associated with that change beyond the name of the destination.

Configuration

By default, properties are configured using the nats.spring prefix:

nats.spring.server specifies the NATS server url or a list of urls in a comma separated list
nats.spring.connection-name the connection name
nats.spring.max-reconnect the maximum reconnects attempts on a single disconnect before the connection closes
nats.spring.reconnect-wait the time, as a duration like 4s, to wait between trying to reconnect to the same server
nats.spring.connection-timeout the time, as a duration like 4s, to wait before cancelling the connection
nats.spring.ping-interval the time, as a duration like 4s, between pings to the server
nats.spring.reconnect-buffer-size the size in bytes for the reconnect buffer
nats.spring.inbox-prefix a custom inbox prefix
nats.spring.no-echo turn off echo from the server
nats.spring.no-no-responders disable no responders support
nats.spring.utf8-support enable UTF-8 subject names (warning this is an experimental feature, not all language clients will support it)
nats.spring.username, nats.spring.password the user name and password to authenticate with
nats.spring.token an authentication token, takes precedence over the username/password
nats.spring.credentials a path to a credentials file, takes precedence over the token and user/pass
nats.spring.nkey a private key (seed) for NKey authentication, takes precedence over the credentials, token and user/pass

TLS can be configured several ways. Set up a default context using system properties like javax.net.ssl.keyStore, set a default SSLContext in the main method before running the spring application, or by setting several properties:

nats.spring.trust-store-password the password for the trust store, defaults to “”
nats.spring.trust-store-path the path to the trust store file
nats.spring.trust-store-provider (optional) the format of the trust store, defaults to “SunX509”. Common options are “SunX509”, “PKIX”.
nats.spring.trust-store-type (optional) the format of the trust store, defaults to “PKCS12”. Common options are “PKCS12”, “JKS”, “JCEKS”, “DKS”.
nats.spring.key-store-password the password for the key store, defaults to “”
nats.spring.key-store-path the path to the key store file
nats.spring.key-store-provider (optional) the format of the key store, defaults to “SunX509”. Common options are “SunX509”, “PKIX”.
nats.spring.key-store-type (optional) the format of the key store, defaults to “PKCS12”. Common options are “PKCS12”, “JKS”, “JCEKS”, “DKS”.
nats.spring.tls-protocol (optional) the preferred protocol for TLS handshake. Common options: TLSv1.2 (default), TLSv1.3.

The key-store-path and trust-store-path must be non-empty to trigger the creation of an SSL context.

Custom Listeners

Custom ConnectionListener and ErrorListeners can be provided by a bean factory. For example, you can implement:

@Bean
public ConnectionListener createConnectionListener(){
    return new ConnectionListener(){
        @Override
        public void connectionEvent(Connection conn,Events type){
            System.out.println("## Custom status change " + type);
        }
    };
}

to create a custom connection listener. The connect-error-sample has an example for both types of listener.

If no custom listeners are provided, a default one is used which will log errors and connection events.

Samples

This repo contains two types of samples. First, there is a stand-alone demo that can be used as starter code, for a POM at least. Second, there is a collection of samples showing the major use-cases implemented by the core code:

autoconfigure-sample a simple command line runner with spring boot that uses the auto-configured nats connection from an application.properties file.
listener-sample uses the binder to listen to a single subject and print messages it receives. The configuration in application.yml specifies a single subject dataIn.
processor-sample creates a processor. The example listens to the subject dataIn and sends to the subject dataOut. If the message contains a UTF-8 string, it is converted to all CAPS before being sent.
polling-sample is similar to the listener sample, but uses polling.
source-sample generates messages on a timer and sends them to a NATS endpoint. In order to leverage this pattern you need to create a named source. See the application.yml for details.
queue-sample listens on a subject and queue group, run multiple copies of the sample to see how messages are load balanced.
multi-connect-sample copy of the processor-sample that uses 2 nats connections.
connect-error-sample demonstrates how to set up a custom connection and error handler.

You can exercise the samples using the nats-sub and nats-pub executables for the client library. For example, to try out the listener:

% java -jar nats-spring-samples/listener-sample/target/listener-sample-0.6.1+3.1.jar --nats.spring.server="nats://localhost:4222"
...
2019-06-24 15:36:43.690  INFO 36282 --- [         nats:3] o.s.cloud.stream.binder.nats.Listener    : received message hello

% nats-pub dataIn "hello"

For the multi-binder, try:

% java -jar nats-spring-samples/processor-sample/target/processor-sample-0.6.1+3.1.jar --nats.spring.server="nats://localhost:4222"
...

% nats-pub dataIn "hello"

 % nats-sub ">"
Listening on [>]
...
[6] Received on [dataIn]: 'hello'
[7] Received on [dataOut]: 'HELLO'

Building This Project

This project is built with maven. The mvnw helper is included in the root folder and its implementation is included in the .mvn folder. You should be able to compile using ./mvnw clean compile, or package the jars with ./mvnw clean package.

Internally there are multiple pom files, one parent for the project, one parent for the samples, one for the autoconfigure code, one for the binder, and one each for the samples. When built, each will have its own artifacts.

Signing and deploying requires that you set up your settings.xml file for maven:

<settings>
    <servers>
        <server>
            <id>ossrh</id>
            <username>xxx</username>
            <password>xxxxx</password>
        </server>
    </servers>
    <profiles>
        <profile>
            <activation>
                <activeByDefault>true</activeByDefault>
            </activation>
            <properties>
                <gpg.keyid>xxx</gpg.keyid>
                <gpg.passphrase>xxxxx</gpg.passphrase>
                <gpg.secretkeyring>xxxxx</gpg.secretkeyring>
            </properties>
        </profile>
    </profiles>
</settings>

Sonatype will accept and close a non-staging repository, but you must manually release it.

Adding New Configuration Properties

All the configuration properties are injected into the NatsProperties class in the autoconfigure module. This class provides a toOptions() method that spits out a valid options configuration from the NATS client library. To add new properties, add them to NatsProperties and update the toOptions() method appropriately.

External Resources

License

Unless otherwise noted, the nats-account-server source files are distributed under the Apache Version 2.0 license found in the LICENSE file.