XUtils

NATS client

NATS client.

Java
https://github.com/nats-io/nats.java

NATS

NATS - Java Client

Simplification

There is a new simplified api that makes working with streams and consumers well, simpler! Simplification is released as of 2.16.14.

Check out the examples:

Service Framework

The service API allows you to easily build NATS services. The Service Framework is released as of 2.16.14

The Services Framework introduces a higher-level API for implementing services with NATS. NATS has always been a strong technology on which to build services, as they are easy to write, are location and DNS independent and can be scaled up or down by simply adding or removing instances of the service.

The Services Framework further streamlines their development by providing observability and standardization. The Service Framework allows your services to be discovered, queried for status and schema information without additional work.

Check out the ServiceExample

Recent Version Notes

Version 2.18.0 (AKA 2.17.7)

2.18.0 attempts to start us on the road to properly Semantic Version (semver). In the last few patch releases, there were technically things that should cause a minor version bump, but were numbered as a patch.

Even if just one api is newly added, semver requires that we bump the minor version. The forceReconnect api is an example of one api being added to the Connection interface. It should have resulted in a minor version bump.

Going forward, when a release contains only bug fixes, it’s appropriate to simply bump the patch. But if an api is added, even one, then the minor version will be bumped.

Force Reconnect

There is are new Connection interface apis:

  • void forceReconnect() throws IOException, InterruptedException;
  • void forceReconnect(ForceReconnectOptions options) throws IOException, InterruptedException;

If you call forceReconnect, your connection will be immediately closed and the reconnect logic will be executed.

Version 2.17.4 Core Improvements

This release was full of core improvements which improve use of more asynchronous behaviors including

  • removing use of synchronized in favor of ReentrantLock
  • The ability to have a dispatcher use an executor to dispatch messages instead of the dispatcher thread being blocking to deliver a message.

Version Notes for older releases

See Version Notes

Downloading the Jar

You can download the latest jar at https://search.maven.org/remotecontent?filepath=io/nats/jnats/2.19.1/jnats-2.19.1.jar.

The examples are available at https://search.maven.org/remotecontent?filepath=io/nats/jnats/2.19.1/jnats-2.19.1-examples.jar.

To use NKeys, you will need the ed25519 library, which can be downloaded at https://repo1.maven.org/maven2/net/i2p/crypto/eddsa/0.3.0/eddsa-0.3.0.jar.

Using Gradle

The NATS client is available in the Maven central repository, and can be imported as a standard dependency in your build.gradle file:

dependencies {
    implementation 'io.nats:jnats:{major.minor.patch}'
}

If you need the latest and greatest before Maven central updates, you can use:

repositories {
    mavenCentral()
    maven {
        url "https://oss.sonatype.org/content/repositories/releases"
    }
}

If you need a snapshot version, you must add the url for the snapshots and change your dependency.

repositories {
    mavenCentral()
    maven {
        url "https://oss.sonatype.org/content/repositories/snapshots"
    }
}

dependencies {
   implementation 'io.nats:jnats:{major.minor.patch}-SNAPSHOT'
}

Using Maven

The NATS client is available on the Maven central repository, and can be imported as a normal dependency in your pom.xml file:

<dependency>
    <groupId>io.nats</groupId>
    <artifactId>jnats</artifactId>
    <version>{major.minor.patch}</version>
</dependency>

If you need the absolute latest, before it propagates to maven central, you can use the repository:

<repositories>
    <repository>
        <id>sonatype releases</id>
        <url>https://oss.sonatype.org/content/repositories/releases</url>
        <releases>
           <enabled>true</enabled>
        </releases>
    </repository>
</repositories>

If you need a snapshot version, you must enable snapshots and change your dependency.

<repositories>
    <repository>
        <id>sonatype snapshots</id>
        <url>https://oss.sonatype.org/content/repositories/snapshots</url>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
</repositories>

<dependency>
    <groupId>io.nats</groupId>
    <artifactId>jnats</artifactId>
    <version>{major.minor.patch}-SNAPSHOT</version>
</dependency>

If you are using the 1.x version of java-nats and don’t want to upgrade to 2.0.0 please use ranges in your POM file, java-nats-streaming 1.x is using [1.1, 1.9.9) for this.

Connecting

There are five different ways to connect using the Java library, each with a parallel method that will allow doing reconnect logic if the initial connect fails. The ability to reconnect on the initial connection failure is NOT an Options setting.

  1. Connect to a local server on the default url. From the Options class: DEFAULT_URL = "nats://localhost:4222";

    // default options 
Connection nc = Nats.connect();


// default options, reconnect on connect 
Connection nc = Nats.connectReconnectOnConnect();

  2. Connect to one or more servers using a URL:

    // single URL, all other default options
Connection nc = Nats.connect("nats://myhost:4222");


// comma-separated list of URLs, all other default options
Connection nc = Nats.connect("nats://myhost:4222,nats://myhost:4223");


// single URL, all other default options, reconnect on connect
Connection nc = Nats.connectReconnectOnConnect("nats://myhost:4222");


// comma-separated list of URLs, all other default options, reconnect on connect
Connection nc = Nats.connectReconnectOnConnect("nats://myhost:4222,nats://myhost:4223");

  3. Connect to one or more servers with a custom configuration:

    Options o = new Options.Builder().server("nats://serverone:4222").server("nats://servertwo:4222").maxReconnects(-1).build();


// custom options
Connection nc = Nats.connect(o);


// custom options, reconnect on connect
Connection nc = Nats.connectReconnectOnConnect(o);

  4. Connect asynchronously, this requires a callback to tell the application when the client is connected:

    Options options = new Options.Builder().server(Options.DEFAULT_URL).connectionListener(handler).build();
Nats.connectAsynchronously(options, true);

  5. Connect with authentication handler:

    AuthHandler authHandler = Nats.credentials(System.getenv("NATS_CREDS"));


// single URL, all other default options
Connection nc = Nats.connect("nats://myhost:4222", authHandler);


// comma-separated list of URLs, all other default options
Connection nc = Nats.connect("nats://myhost:4222,nats://myhost:4223", authHandler);


// single URL, all other default options, reconnect on connect
Connection nc = Nats.connectReconnectOnConnect("nats://myhost:4222", authHandler);


// comma-separated list of URLs, all other default options, reconnect on connect
Connection nc = Nats.connectReconnectOnConnect("nats://myhost:4222,nats://myhost:4223", authHandler);

Clusters & Reconnecting

The Java client will automatically reconnect if it loses its connection the nats-server. If given a single server, the client will keep trying that one. If given a list of servers, the client will rotate between them. When the nats servers are in a cluster, they will tell the client about the other servers, so that in the simplest case a client could connect to one server, learn about the cluster and reconnect to another server if its initial one goes down.

To tell the connection about multiple servers for the initial connection, use the servers() method on the options builder, or call server() multiple times.

String[] serverUrls = {"nats://serverOne:4222", "nats://serverTwo:4222"};
Options o = new Options.Builder().servers(serverUrls).build();

Reconnection behavior is controlled via a few options, see the javadoc for the Options.Builder class for specifics on reconnect limits, delays and buffers.

Connection Options

Connection options are configured using the Options class. There is a Builder which uses a fluent interface. It can accept Properties object or a path to a Properties file.

Property Names

The io.nats.client. prefix is not required in the properties file anymore. These are now equivalent:

io.nats.client.servers=nats://localhost:4222

servers=nats://localhost:4222

Last one wins

The Options builder allows you to use both properties and code. When it comes to the builder, the last one called wins. This applies to each individual property.

props.setProperty(Options.PROP_MAX_MESSAGES_IN_OUTGOING_QUEUE, "7000");

o = new Options.Builder()
   .properties(props)
   .maxMessagesInOutgoingQueue(6000)
   .build();
assertEquals(6000, o.getMaxMessagesInOutgoingQueue());

o = new Options.Builder()
   .maxMessagesInOutgoingQueue(6000)
   .properties(props)
   .build();
assertEquals(7000, o.getMaxMessagesInOutgoingQueue());

o = new Options.Builder()
    .maxMessagesInOutgoingQueue(6000)
    .maxMessagesInOutgoingQueue(8000)
    .build();
assertEquals(8000, o.getMaxMessagesInOutgoingQueue());

AuthHandler and JWT Credentials

You can manually create the AuthHandler and set it in the options

AuthHandler ah = Nats.credentials("path/to/my.creds");
Options options = new Options.Builder()
    .authHandler(ah)
    .build();

or you can now set the file path directly and an AuthHandler will be created:

Options options = new Options.Builder()
    .credentialPath("path/to/my.creds")
    .build();

The developer can also set the credential path in a properties file:

io.nats.client.credential.path=path/to/my.creds

SSLContext

The Options builder has several options set use or affect creation of an SSLContext

// Provide the SSLContext
public Builder sslContext(SSLContext ctx)

// Generic SSLContext Creation
public Builder secure()
public Builder opentls()

// Custom SSLContext Creation Properties
public Builder keystore(String keystore)
public Builder keystorePassword(char[] keystorePassword)
public Builder truststore(String truststore)
public Builder truststorePassword(char[] truststorePassword)
public Builder tlsAlgorithm(String tlsAlgorithm)

There are equivalent properties for these builder methods (except sslContext):

# Generic SSLContext Creation
io.nats.client.secure=true
io.nats.client.opentls=true

# Custom SSLContext Creation Properties
io.nats.client.keyStore=path/to/keystore.jks
io.nats.client.keyStorePassword=kspassword
io.nats.client.trustStore=path/to/truststore.jks
io.nats.client.trustStorePassword=tspassword
io.nats.client.tls.algorithm=SunX509

When options are built, the SSLContext will be accepted or created in the following order.

  1. If it’s directly provided via the builder sslContext(SSLContext ctx) method.
  2. If keyStore is provided, an SSLContext will be created using all custom properties. If not supplied, the tls algorithm is SunX509
  3. If opentls is true or any of the bootstrap servers has opentls as their scheme, a generic SSLContext will be created that “trusts all certs”.
  4. If secure is true or any of the bootstrap servers has tls or wss as their scheme, the javax.net.ssl.SSLContext.getDefault() will be used.

Publishing

Once connected, publishing is accomplished via one of three methods:

  1. With a subject and message body:

    nc.publish("subject", "hello world".getBytes(StandardCharsets.UTF_8));

  2. With a subject and message body, as well as a subject for the receiver to reply to:

    nc.publish("subject", "replyto", "hello world".getBytes(StandardCharsets.UTF_8));

  3. As a request that expects a reply. This method uses a Future to allow the application code to wait for the response. Under the covers a request/reply pair is the same as a publish/subscribe only the library manages the subscription for you.

    Future<Message> incoming = nc.request("subject", "hello world".getBytes(StandardCharsets.UTF_8));
Message msg = incoming.get(500, TimeUnit.MILLISECONDS);
String response = new String(msg.getData(), StandardCharsets.UTF_8);

All of these methods, as well as the incoming message code use byte arrays for maximum flexibility. Applications can send JSON, Strings, YAML, Protocol Buffers, or any other format through NATS to applications written in a wide range of languages.

ReplyTo When Making A Request

The Message object allows you to set a replyTo, but in requests, the replyTo is reserved for internal use as the address for the server to respond to the client with the consumer’s reply.

Listening for Incoming Messages

The Java NATS library provides two mechanisms to listen for messages, three if you include the request/reply discussed above.

  1. Synchronous subscriptions where the application code manually asks for messages and blocks until they arrive. Each subscription is associated with a single subject, although that subject can be a wildcard.

    Subscription sub = nc.subscribe("subject");
Message msg = sub.nextMessage(Duration.ofMillis(500));


String response = new String(msg.getData(), StandardCharsets.UTF_8);

  2. A Dispatcher that will call application code in a background thread. Dispatchers can manage multiple subjects with a single thread and shared callback.

    Dispatcher d = nc.createDispatcher((msg) -> {
    String response = new String(msg.getData(), StandardCharsets.UTF_8);
    ...
});


d.subscribe("subject");

    A dispatcher can also accept individual callbacks for any given subscription.

    Dispatcher d = nc.createDispatcher((msg) -> {});


Subscription s = d.subscribe("some.subject", (msg) -> {
    String response = new String(msg.getData(), StandardCharsets.UTF_8);
    System.out.println("Message received (up to 100 times): " + response);
});
d.unsubscribe(s, 100);

JetStream

Publishing and subscribing to JetStream-enabled servers is straightforward. A JetStream-enabled application will connect to a server, establish a JetStream context, and then publish or subscribe. This can be mixed and matched with standard NATS subject, and JetStream subscribers, depending on configuration, receive messages from both streams and directly from other NATS producers.

The JetStream Context

After establishing a connection as described above, create a JetStream Context.

JetStream js = nc.jetStream();

You can pass options to configure the JetStream client, although the defaults should suffice for most users. See the JetStreamOptions class.

There is no limit to the number of contexts used, although normally one would only require a single context. Contexts may be prefixed to be used in conjunction with NATS authorization.

Subject and Queue Name Validation

For subjects, the client was strict when validating subject names for consumer subject filters and subscriptions. It only allowed printable ascii characters except for *, >, ., \\ and /. This restriction has been changed to the following:

  • cannot contain spaces \r \n \t
  • cannot start or end with subject token delimiter .
  • cannot have empty segments

This means that UTF characters are now allowed in subjects in this client.

For queue names, there has been inconsistent validation, if any. Queue names now require the same validation as subjects.

Subscribe Subject Validation

Additionally, for subjects used in subscribe api, applications may start throwing an exception:

90011 Subject does not match consumer configuration filter

Let’s say you have a stream with subject foo.> And you are subscribing to foo.a. When you don’t supply a filter subject on a consumer, it becomes >, which means all subjects.

So this is a problem, because you think you are subscribing to foo.a but in reality, without this check, you will be getting all messages foo.> subjects, not just foo.a

Validating the subscribe subject against the filter subject is needed to prevent this. Unfortunately, this makes existing code throw the 90011 exception.

Publishing

To publish messages, use the JetStream.publish(...) API. A stream must be established before publishing. You can publish in either a synchronous or asynchronous manner.

Synchronous

// create a typical NATS message
Message msg = NatsMessage.builder()
   .subject("foo")
   .data("hello", StandardCharsets.UTF_8)
   .build();

PublishAck pa = js.publish(msg);

See NatsJsPub.java in the JetStream examples for a detailed and runnable example.

If there is a problem an exception will be thrown, and the message may not have been persisted. Otherwise, the stream name and sequence number is returned in the publish acknowledgement.

There are a variety of publish options that can be set when publishing. When duplicate checking has been enabled on the stream, a message ID should be set. One set of options are expectations. You can set a publish expectation such as a particular stream name, previous message ID, or previous sequence number. These are hints to the server that it should reject messages where these are not met, primarily for enforcing your ordering or ensuring messages are not stored on the wrong stream.

The PublishOptions are immutable, but the builder an be re-used for expectations by clearing the expected.

For example:

PublishOptions.Builder pubOptsBuilder = PublishOptions.builder()
        .expectedStream("TEST")
        .messageId("mid1");
PublishAck pa = js.publish("foo", null, pubOptsBuilder.build());

pubOptsBuilder.clearExpected()
        .setExpectedLastMsgId("mid1")
        .setExpectedLastSequence(1)
        .messageId("mid2");
pa = js.publish("foo", null, pubOptsBuilder.build());

See NatsJsPubWithOptionsUseCases.java in the JetStream examples for a detailed and runnable example.

Asynchronous

List<CompletableFuture<PublishAck>> futures = new ArrayList<>();
for (int x = 1; x < roundCount; x++) {
    // create a typical NATS message
    Message msg = NatsMessage.builder()
    .subject("foo")
    .data("hello", StandardCharsets.UTF_8)
    .build();

    // Publish a message
    futures.add(js.publishAsync(msg));
}

for (CompletableFuture<PublishAck> future : futures) {
   ... process the futures
}

See the NatsJsPubAsync.java in the JetStream examples for a detailed and runnable example.

ReplyTo When Publishing

The Message object allows you to set a replyTo, but in publish requests, the replyTo is reserved for internal use as the address for the server to respond to the client with the PublishAck.

Consuming Messages

There are two methods of subscribing, Push and Pull with each variety having its own set of options and abilities.

Asynchronous Push Subscription

Push subscriptions can be synchronous or asynchronous. The server pushes messages to the client.

Dispatcher disp = ...;

MessageHandler handler = (msg) -> {
// Process the message.
// Ack the message depending on the ack model
};

PushSubscribeOptions so = PushSubscribeOptions.builder()
   .durable("optional-durable-name")
   .build();

boolean autoAck = ...

js.subscribe("my-subject", disp, handler, autoAck);

See the NatsJsPushSubWithHandler.java in the JetStream examples for a detailed and runnable example.

Subscribe Subject

With the introduction of simplification, the various original subscribe methods available will eventually be deprecated. But since they are available, we need to address the concept of the subscribe subject.

Consider the example:

js.subscribe("my-subject", disp, handler, autoAck);

The subscribe method takes a “subject” as the first parameter. We call this the subscribe subject. The subject could be something like my.subject or my.star.* or my.gt.> or even >. This parameter is used and validated in different ways depending on the context of the call, including looking up the stream, if stream is not provided via subscribe options.

The subscribe subject could be used to make a simple subscription. In this case it is required. An ephemeral consumer will be created for that subject, assuming that subject can be looked up in a stream.

JetStream js = nc.jetStream();
JetStreamSubscription sub = subscribe(subject)

If subscribe call has either a PushSubscribeOptions or PullSubscribeOptions that have a ConsumerConfiguration with one or more filter subjects, the subscribe subject is optional since we can use the first filter subject as the subscribe subject.

PushSubscribeOptions pso = ConsumerConfiguration.builder().filterSubject("my.subject").buildPushSubscribeOptions();
js.subscribe(null, pso);

The other time you can skip the subject parameter is when you “bind”. Since the stream name and consumer name are part of the bind, the subject will be retrieved from the consumer looked-up via the bind stream and consumer name information.

Synchronous Consuming

See NatsJsPushSub.java in the JetStream examples for a detailed and runnable example.

PushSubscribeOptions so = PushSubscribeOptions.builder()
        .durable("optional-durable-name")
        .build();

// Subscribe synchronously, then just wait for messages.
JetStreamSubscription sub = js.subscribe("subject", so);
nc.flush(Duration.ofSeconds(5));

Message msg = sub.nextMessage(Duration.ofSeconds(1));

Multiple Filter Subjects

The client has the ability to have multiple filter subjects for any single JetStream consumer. This can only be set up via the Consumer Configuration.

ConsumerConfiguration cc = ConsumerConfiguration.builder()
    ...
    .filterSubjects("subject1", "subject2")
    .build();

Message Acknowledgements

There are multiple types of acknowledgements in JetStream:

  • Message.ack(): Acknowledges a message.
  • Message.ackSync(Duration): Acknowledges a message and waits for a confirmation. When used with deduplications this creates exactly once delivery guarantees (within the deduplication window). This may significantly impact performance of the system.
  • Message.nak(): A negative acknowledgment indicating processing failed and the message should be resent later.
  • Message.term(): Never send this message again, regardless of configuration.
  • Message.inProgress(): The message is being processed and reset the redelivery timer in the server. The message must be acknowledged later when processing is complete.

Note that exactly once delivery guarantee can be achieved by using a consumer with explicit ack mode attached to stream setup with a deduplication window and using the ackSync to acknowledge messages. The guarantee is only valid for the duration of the deduplication window.

Pull Subscriptions

Pull subscriptions are always synchronous. The server organizes messages into a batch which it sends when requested.

PullSubscribeOptions pullOptions = PullSubscribeOptions.builder()
   .durable("durable-name-is-optional")
   .build();
JetStreamSubscription sub = js.subscribe("subject", pullOptions);

Bind

Pull subscriptions allow for binding to existing consumers. The best practice is to provide null for the subscribe subject, but if you do provide it, it must match the consumer subject filter, or you will receive an IllegalArgumentException. See client errors below and JsSubSubjectDoesNotMatchFilter 90011

  1. Short Form

    PullSubscribeOptions pullOptions = PullSubscribeOptions.bind("stream", "durable-name");
JetStreamSubscription sub = js.subscribe(null, pullOptions);

  2. Long Form

    PullSubscribeOptions pullOptions = PullSubscribeOptions.builder()
    .stream("stream")
    .durable("durable-name")
    .bind(true)
    .build();

Fetch

List<Message> message = sub.fetch(100, Duration.ofSeconds(1));
for (Message m : messages) {
   // process message
   m.ack();
}

The fetch method is a macro pull that uses advanced pulls under the covers to return a list of messages. The list may be empty or contain at most the batch size. All status messages are handled for you except terminal status messages. See Pull Exception Handling below. The client can provide a timeout to wait for the first message in a batch. The fetch call returns when the batch is ready. If the timeout is exceeded while messages are in flight, but before they reach the client, those messages will be available via nextMessage or will be used to fulfill the next fetch.

One important thing to consider when using this is ack wait. Once the server sends a message, it’s specific ack wait timer is started. If you ask for too many messages, you may fail to ack all messages in time and can get redeliveries.

See NatsJsPullSubFetch.java and NatsJsPullSubFetchUseCases.java in the JetStream examples for a detailed and runnable example.

Iterate

        Iterator<Message> iter = sub.iterate(100, Duration.ofSeconds(1));
        while (iter.hasNext()) {
            Message m = iter.next();
            // process message
            m.ack();
        }

The iterate method is a macro pull that uses advanced pulls under the covers to return an iterator. The iterator may have no messages up to at most the batch size. All status messages are handled for you except terminal status messages. See Pull Exception Handling below. The client can provide a timeout to wait for the first message in a batch. The iterate call returns the iterator immediately, but under the covers it will wait for the first message based on the timeout.

The iterate method is usually preferred to the fetch method as it allows you to start processing messages right away instead of waiting until the entire batch is filled. This reduces problems with ack wait and generally is more efficient.

See NatsJsPullSubIterate.java and NatsJsPullSubIterateUseCases.java in the JetStream examples for a detailed and runnable example.

Batch Size

sub.pull(100);
...
Message m = sub.nextMessage(Duration.ofSeconds(1));

This is an advanced / raw pull that specifies a batch size. When asked, the server will send whatever messages it has up to the batch size. If it has no messages it will wait until it has some to send. The pull request only completes on the server once the entire batch has been sent. It’s up to you to track this and only send pulls when the batch is complete, or you risk having pulls stack up and possibly receiving a status 409 Exceeded MaxWaiting warning. The nextMessage request may time out but that does not indicate that there are no more messages in the pull. Instead, it indicates that there is no message available at that time. Once the entire batch size has been filled, you must make another pull request.

See NatsJsPullSubBatchSize.java and NatsJsPullSubBatchSizeUseCases.java in the JetStream examples for detailed and runnable example.

No Wait and Batch Size

sub.pullNoWait(100);
...
Message m = sub.nextMessage(Duration.ofSeconds(1));

This is an advanced / raw pull that also specifies a batch size. When asked, the server will send whatever messages it has at the moment the pull request is processed by the server, up to the batch size. If there are less than the batch size available, you will get what is available. You must make a pull request every time.

See the NatsJsPullSubNoWaitUseCases.java in the JetStream examples for a detailed and runnable example.

Expires In and Batch Size

sub.pullExpiresIn(100, Duration.ofSeconds(3));
...
Message m = sub.nextMessage(Duration.ofSeconds(4));

Another advanced version of pull specifies a maximum time to wait for the batch to fill. The server sends messages up until the batch is filled or the time expires. It’s important to set your client’s nextMessage timeout to be longer than the time you’ve asked the server to expire in. Once nextMessage returns null, you know your pull is done, and you can make another one.

See NatsJsPullSubExpire.java and NatsJsPullSubExpireUseCases.java in the JetStream examples for detailed and runnable examples.

Pull Exception Handling

Ordered Push Subscription Option

You can now set a Push Subscription option called “Ordered”. When you set this flag, library will take over creation of the consumer and create a subscription that guarantees the order of messages. This consumer will use flow control with a default heartbeat of 5 seconds. Messages will not require acks as the Ack Policy will be set to No Ack. When creating the subscription, there are some restrictions for the consumer configuration settings.

  • Ack policy must be AckPolicy.None (or left un-set). maxAckPending will be ignored.
  • Deliver Group (aka Queue) cannot be used
  • You cannot set a durable consumer name
  • You cannot set the deliver subject
  • max deliver can only be set to 1 (or left un-set)
  • The idle heartbeat cannot be less than 5 seconds. Flow control will automatically be used.

You can however set the deliver policy which will be used to start the subscription.

Client Error Messages

`In addition to some generic validation messages for values in builders, there are also additional grouped and numbered client error messages:

  • Subscription building and creation
  • Consumer creation
  • Object Store operations
| Name                                         | Group-Code| Description                                                                                                    |
|----------------------------------------------|-----------|----------------------------------------------------------------------------------------------------------------|
| JsSoDurableMismatch                          | SO-90101  | Builder durable must match the consumer configuration durable if both are provided.                            |
| JsSoDeliverGroupMismatch                     | SO-90102  | Builder deliver group must match the consumer configuration deliver group if both are provided.                |
| JsSoDeliverSubjectMismatch                   | SO-90103  | Builder deliver subject must match the consumer configuration deliver subject if both are provided.            |
| JsSoOrderedNotAllowedWithBind                | SO-90104  | Bind is not allowed with an ordered consumer.                                                                  |
| JsSoOrderedNotAllowedWithDeliverGroup        | SO-90105  | Deliver group is not allowed with an ordered consumer.                                                         |
| JsSoOrderedNotAllowedWithDurable             | SO-90106  | Durable is not allowed with an ordered consumer.                                                               |
| JsSoOrderedNotAllowedWithDeliverSubject      | SO-90107  | Deliver subject is not allowed with an ordered consumer.                                                       |
| JsSoOrderedRequiresAckPolicyNone             | SO-90108  | Ordered consumer requires Ack Policy None.                                                                     |
| JsSoOrderedRequiresMaxDeliverOfOne           | SO-90109  | Max deliver is limited to 1 with an ordered consumer.                                                          |
| JsSoNameMismatch                             | SO-90110  | Builder name must match the consumer configuration name if both are provided.                                  |
| JsSoOrderedMemStorageNotSuppliedOrTrue       | SO-90111  | Mem Storage must be true if supplied.                                                                          |
| JsSoOrderedReplicasNotSuppliedOrOne          | SO-90112  | Replicas must be 1 if supplied.                                                                                |
| JsSoNameOrDurableRequiredForBind             | SO-90113  | Name or Durable required for Bind.                                                                             |
| JsSubPullCantHaveDeliverGroup                | SUB-90001 | Pull subscriptions can't have a deliver group.                                                                 |
| JsSubPullCantHaveDeliverSubject              | SUB-90002 | Pull subscriptions can't have a deliver subject.                                                               |
| JsSubPushCantHaveMaxPullWaiting              | SUB-90003 | Push subscriptions cannot supply max pull waiting.                                                             |
| JsSubQueueDeliverGroupMismatch               | SUB-90004 | Queue / deliver group mismatch.                                                                                |
| JsSubFcHbNotValidPull                        | SUB-90005 | Flow Control and/or heartbeat is not valid with a pull subscription.                                           |
| JsSubFcHbNotValidQueue                       | SUB-90006 | Flow Control and/or heartbeat is not valid in queue mode.                                                      |
| JsSubNoMatchingStreamForSubject              | SUB-90007 | No matching streams for subject.                                                                               |
| JsSubConsumerAlreadyConfiguredAsPush         | SUB-90008 | Consumer is already configured as a push consumer.                                                             |
| JsSubConsumerAlreadyConfiguredAsPull         | SUB-90009 | Consumer is already configured as a pull consumer.                                                             |
| _removed_                                    | SUB-90010 |                                                                                                                |
| JsSubSubjectDoesNotMatchFilter               | SUB-90011 | Subject does not match consumer configuration filter.                                                          |
| JsSubConsumerAlreadyBound                    | SUB-90012 | Consumer is already bound to a subscription.                                                                   |
| JsSubExistingConsumerNotQueue                | SUB-90013 | Existing consumer is not configured as a queue / deliver group.                                                |
| JsSubExistingConsumerIsQueue                 | SUB-90014 | Existing consumer is configured as a queue / deliver group.                                                    |
| JsSubExistingQueueDoesNotMatchRequestedQueue | SUB-90015 | Existing consumer deliver group does not match requested queue / deliver group.                                |
| JsSubExistingConsumerCannotBeModified        | SUB-90016 | Existing consumer cannot be modified.                                                                          |
| JsSubConsumerNotFoundRequiredInBind          | SUB-90017 | Consumer not found, required in bind mode.                                                                     |
| JsSubOrderedNotAllowOnQueues                 | SUB-90018 | Ordered consumer not allowed on queues.                                                                        |
| JsSubPushCantHaveMaxBatch                    | SUB-90019 | Push subscriptions cannot supply max batch.                                                                    |
| JsSubPushCantHaveMaxBytes                    | SUB-90020 | Push subscriptions cannot supply max bytes.                                                                    |
| JsSubPushAsyncCantSetPending                 | SUB-90021 | Pending limits must be set directly on the dispatcher.                                                         |
| JsSubSubjectNeededToLookupStream             | SUB-90022 | Subject needed to lookup stream. Provide either a subscribe subject or a ConsumerConfiguration filter subject. |
| JsConsumerCreate290NotAvailable              | CON-90301 | Name field not valid when v2.9.0 consumer create api is not available.                                         |
| JsConsumerNameDurableMismatch                | CON-90302 | Name must match durable if both are supplied.                                                                  |
| JsMultipleFilterSubjects210NotAvailable      | CON-90303 | Multiple filter subjects not available until server version 2.10.0.                                            |
| OsObjectNotFound                             | OS-90201  | The object was not found.                                                                                      |
| OsObjectIsDeleted                            | OS-90202  | The object is deleted.                                                                                         |
| OsObjectAlreadyExists                        | OS-90203  | An object with that name already exists.                                                                       |
| OsCantLinkToLink                             | OS-90204  | A link cannot link to another link.                                                                            |
| OsGetDigestMismatch                          | OS-90205  | Digest does not match meta data.                                                                               |
| OsGetChunksMismatch                          | OS-90206  | Number of chunks does not match meta data.                                                                     |
| OsGetSizeMismatch                            | OS-90207  | Total size does not match meta data.                                                                           |
| OsGetLinkToBucket                            | OS-90208  | Cannot get object, it is a link to a bucket.                                                                   |
| OsLinkNotAllowOnPut                          | OS-90209  | Link not allowed in metadata when putting an object.                                                           |

TLS Certs

The raw TLS test certs are in src/test/resources/certs and come from the nats.go repository. However, the java client also needs a keystore and truststore.jks files for creating a context. These can be created using:

> cd src/test/resources
> keytool -keystore truststore.jks -alias CARoot -import -file certs/ca.pem -storepass password -noprompt -storetype pkcs12
> cat certs/client-key.pem certs/client-cert.pem > combined.pem
> openssl pkcs12 -export -in combined.pem -out cert.p12
> keytool -importkeystore -srckeystore cert.p12 -srcstoretype pkcs12 -deststoretype pkcs12 -destkeystore keystore.jks
> keytool -keystore keystore.jks -alias CARoot -import -file certs/ca.pem -storepass password -noprompt
> rm cert.p12 combined.pem

TLS Handshake First

In Server 2.10.3 and later, there is the ability to have TLS Handshake First.

The server config will contain this:

tls {
  ...
  handshake_first: 300ms
}

TLS Handshake First is used to instruct the library perform the TLS handshake right after the connect and before receiving the INFO protocol from the server. If this option is enabled but the server is not configured to perform the TLS handshake first, the connection will fail.

Reverse Proxy

In a reverse proxy configuration, the client connects securely to the reverse proxy and the proxy may connect securely or insecurely to the server.

If the proxy connects securely to the server, then there is nothing special required to do at all.

But most commonly, the proxy connects insecurely to the server. This is where server configuration comes into play. You will need to configure the server like so:

tls {}
allow_non_tls: true

Before this, the client would not connect because the server was not requiring tls for the proxy, but the client was configured as secure because it was connecting securely to the proxy. The client thought that this was a mismatch and would not connect, essentially failing fast instead of waiting for the server to reject the connection attempt.

The latest version of the client is able to recognize this server configuration and understands that it’s okay to connect securely to the proxy regardless of the server configuration.

You just have to make sure you can properly connect securely to the proxy and that’s where the code in this sample comes in.

SSL/TLS Performance

After recent tests we realized that TLS performance is lower than we would like. After researching the problem and possible solutions we came to a few conclusions:

  • TLS performance for the native JDK has not been historically great
  • TLS performance is better in JDK12 than JDK8
  • A small fix to the library in 2.5.1 allows the use of https://github.com/google/conscrypt and https://github.com/wildfly/wildfly-openssl, conscrypt provides the best performance in our tests
  • TLS still comes at a price (1gb/s vs 4gb/s in some tests), but using the JNI libraries can result in a 10x boost in our testing
  • If TLS performance is reasonable for your application we recommend using the j2se implementation for simplicity

To use conscrypt or wildfly, you will need to add the appropriate jars to your class path and create an SSL context manually. This context can be passed to the Options used when creating a connection. The NATSAutoBench example provides a conscrypt flag which can be used to try out the library, manually including the jar is required.

Version Notes

Version 2.17.2 Message Immutability Headers Bug

Once a message is created, it is intended to be immutable. Before 2.17.2, the Headers object could be modified (via put, add, remove) after construction, either directly with the developer’s original Headers object or from the one available via Message.getHeaders(). This will cause a protocol failure when the message is written to the server, because the protocol size had already been calculated. This calculation is done at construction time because there are multiple places in the workflow that rely on the protocol size, so it must not change once created.

Version 2.16.0: Consumer Create

This release by default will use a new JetStream consumer create API when interacting with nats-server version 2.9.0 or higher. This changes the subjects used by the client to create consumers, which might in some cases require changes in access and import/export configuration. The developer can opt out of using this feature by using a custom JetStreamOptions and using it when creating JetStream, Key Value and Object Store regular and management contexts.

JetStreamOptions jso = JetStreamOptions.builder().optOut290ConsumerCreate(true).build();

JetStream js = connection.jetStream(jso);
JetStreamManagement jsm = connection.jetStreamManagement(jso);
KeyValue kv = connection.keyValue("bucket", KeyValueOptions.builder(jso).build());
KeyValueManagement kvm = connection.keyValueManagement(KeyValueOptions.builder(jso).build());
ObjectStore os = connection.objectStore("bucket", ObjectStoreOptions.builder(jso).build());
ObjectStoreManagement osm = connection.objectStoreManagement(ObjectStoreOptions.builder(jso).build());

Building From Source

The build depends on Gradle, and contains gradlew to simplify the process. After cloning, you can build the repository and run the tests with a single command:

> git clone https://github.com/nats-io/nats.java
> cd nats.java
> ./gradlew clean build

Or to build without tests

> ./gradlew clean build -x test

This will place the class files in a new build folder. To just build the jar:

> ./gradlew jar

The jar will be placed in build/libs.

You can also build the java doc, and the samples jar using:

> ./gradlew javadoc
> ./gradlew exampleJar

The java doc is located in build/docs and the example jar is in build/libs. Finally, to run the tests with the coverage report:

> ./gradlew test jacocoTestReport

which will create a folder called build/reports/jacoco containing the file index.html you can open and use to browse the coverage. Keep in mind we have focused on library test coverage, not coverage for the examples.

Many of the tests run nats-server on a custom port. If nats-server is in your path they should just work, but in cases where it is not, or an IDE running tests has issues with the path you can specify the nats-server location with the environment variable nats_server_path.

Articles

  • coming soon...