Merge pull request #1314 from nats-io/ordered-consumer-nameable

Ability to supply a prefix for watches. Fixed ordered consumer naming bug.

Author: scottf

src/main/java/io/nats/client/BaseConsumeOptions.java

@@ -20,10 +20,10 @@
 
 import static io.nats.client.support.ApiConstants.*;
 import static io.nats.client.support.JsonUtils.*;
+import static io.nats.client.support.JsonValueUtils.*;
 import static io.nats.client.support.JsonValueUtils.readBoolean;
 import static io.nats.client.support.JsonValueUtils.readInteger;
 import static io.nats.client.support.JsonValueUtils.readLong;
-import static io.nats.client.support.JsonValueUtils.*;
 
 /**
  * Base Consume Options are provided to customize the way the consume and
@@ -208,7 +208,7 @@ public B thresholdPercent(int thresholdPercent) {
 
         /**
          * Raise status warning turns on sending status messages to the error listener.
-         * The default of to not raise status warning
+         * The default is to not raise status warnings
          * @return the builder
          */
         public B raiseStatusWarnings() {
@@ -218,7 +218,8 @@ public B raiseStatusWarnings() {
 
         /**
          * Turn on or off raise status warning turns. When on, status messages are sent to the error listener.
-         * The default of to not raise status warning
+         * The default is to not raise status warnings
+         * @param raiseStatusWarnings flag indicating whether to raise status messages
          * @return the builder
          */
         public B raiseStatusWarnings(boolean raiseStatusWarnings) {

src/main/java/io/nats/client/BaseConsumerContext.java

@@ -31,8 +31,8 @@ public interface BaseConsumerContext {
      * Read the next message with max wait set to {@value BaseConsumeOptions#DEFAULT_EXPIRES_IN_MILLIS} ms
      * @return the next message or null if the max wait expires
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
-     * @throws InterruptedException if one is thrown, in order to propagate it up
+     *         server, such as timeout or interruption
+     * @throws InterruptedException if one is thrown, to propagate it up
      * @throws JetStreamStatusCheckedException an exception representing a status that requires attention,
      *         such as the consumer was deleted on the server in the middle of use.
      * @throws JetStreamApiException the request had an error related to the data
@@ -44,8 +44,8 @@ public interface BaseConsumerContext {
      * @param maxWait duration of max wait. Cannot be less than {@value BaseConsumeOptions#MIN_EXPIRES_MILLS} milliseconds.
      * @return the next message or null if the max wait expires
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
-     * @throws InterruptedException if one is thrown, in order to propagate it up
+     *         server, such as timeout or interruption
+     * @throws InterruptedException if one is thrown, to propagate it up
      * @throws JetStreamStatusCheckedException an exception representing a status that requires attention,
      *         such as the consumer was deleted on the server in the middle of use.
      * @throws JetStreamApiException the request had an error related to the data
@@ -57,8 +57,8 @@ public interface BaseConsumerContext {
      * @param maxWaitMillis the max wait value in milliseconds. Cannot be less than {@value BaseConsumeOptions#MIN_EXPIRES_MILLS} milliseconds.
      * @return the next message or null if the max wait expires
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
-     * @throws InterruptedException if one is thrown, in order to propagate it up
+     *         server, such as timeout or interruption
+     * @throws InterruptedException if one is thrown, to propagate it up
      * @throws JetStreamStatusCheckedException an exception representing a status that requires attention,
      *         such as the consumer was deleted on the server in the middle of use.
      * @throws JetStreamApiException the request had an error related to the data
@@ -67,10 +67,10 @@ public interface BaseConsumerContext {
 
     /**
      * Start a one use Fetch Consumer using all defaults other than the number of messages. See {@link FetchConsumer}
-     * @param maxMessages the maximum number of message to consume
+     * @param maxMessages the maximum number of messages to consume
      * @return the FetchConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     FetchConsumer fetchMessages(int maxMessages) throws IOException, JetStreamApiException;
@@ -80,38 +80,38 @@ public interface BaseConsumerContext {
      * @param maxBytes the maximum number of bytes to consume
      * @return the FetchConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     FetchConsumer fetchBytes(int maxBytes) throws IOException, JetStreamApiException;
 
     /**
-     * Start a one use Fetch Consumer with complete custom consume options. See {@link FetchConsumer}
+     * Start a one-use Fetch Consumer with complete custom consume options. See {@link FetchConsumer}
      * @param fetchConsumeOptions the custom fetch consume options. See {@link FetchConsumeOptions}
      * @return the FetchConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     FetchConsumer fetch(FetchConsumeOptions fetchConsumeOptions) throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running IterableConsumer with default ConsumeOptions. See {@link IterableConsumer} and {@link ConsumeOptions}
-     * IterableConsumer require the developer call nextMessage.
+     * IterableConsumer require the developer calls nextMessage.
      * @return the IterableConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     IterableConsumer iterate() throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running IterableConsumer with custom ConsumeOptions. See {@link IterableConsumer} and {@link ConsumeOptions}
-     * IterableConsumer requires the developer call nextMessage.
+     * IterableConsumer requires the developer calls nextMessage.
      * @param consumeOptions the custom consume options
      * @return the IterableConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     IterableConsumer iterate(ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
@@ -121,7 +121,7 @@ public interface BaseConsumerContext {
      * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(MessageHandler handler) throws IOException, JetStreamApiException;
@@ -130,10 +130,10 @@ public interface BaseConsumerContext {
      * Start a long-running MessageConsumer with default ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
      *
      * @param dispatcher The dispatcher to handle this subscription
-     * @param handler    the MessageHandler used for receiving messages.
+     * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
-     * @throws IOException           covers various communication issues with the NATS
-     *                               server such as timeout or interruption
+     * @throws IOException covers various communication issues with the NATS
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(Dispatcher dispatcher, MessageHandler handler) throws IOException, JetStreamApiException;
@@ -142,10 +142,10 @@ public interface BaseConsumerContext {
      * Start a long-running MessageConsumer with custom ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
      *
      * @param consumeOptions the custom consume options
-     * @param handler        the MessageHandler used for receiving messages.
+     * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
-     * @throws IOException           covers various communication issues with the NATS
-     *                               server such as timeout or interruption
+     * @throws IOException covers various communication issues with the NATS
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(ConsumeOptions consumeOptions, MessageHandler handler) throws IOException, JetStreamApiException;
@@ -154,11 +154,11 @@ public interface BaseConsumerContext {
      * Start a long-running MessageConsumer with custom ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
      *
      * @param consumeOptions the custom consume options
-     * @param dispatcher     The dispatcher to handle this subscription
-     * @param handler        the MessageHandler used for receiving messages.
+     * @param dispatcher the dispatcher to handle this subscription
+     * @param handler the MessageHandler used for receiving messages.
      * @return the MessageConsumer instance
-     * @throws IOException           covers various communication issues with the NATS
-     *                               server such as timeout or interruption
+     * @throws IOException covers various communication issues with the NATS
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     MessageConsumer consume(ConsumeOptions consumeOptions, Dispatcher dispatcher, MessageHandler handler) throws IOException, JetStreamApiException;

src/main/java/io/nats/client/KeyValueManagement.java

@@ -28,7 +28,7 @@ public interface KeyValueManagement {
      * @param config the key value configuration
      * @return bucket info
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      * @throws IllegalArgumentException the server is not JetStream enabled
      */
@@ -39,7 +39,7 @@ public interface KeyValueManagement {
      * @param config the key value configuration
      * @return bucket info
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      * @throws IllegalArgumentException the server is not JetStream enabled
      */
@@ -49,7 +49,7 @@ public interface KeyValueManagement {
      * Get the list of bucket names.
      * @return list of bucket names
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *          server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     List<String> getBucketNames() throws IOException, JetStreamApiException;
@@ -59,7 +59,7 @@ public interface KeyValueManagement {
      * @deprecated Use {@link #getStatus(String)} instead.
      * @param bucketName the bucket name to use
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      * @return the bucket status object
      */
@@ -70,7 +70,7 @@ public interface KeyValueManagement {
      * Gets the status for an existing bucket.
      * @param bucketName the bucket name to use
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      * @return the bucket status object
      */
@@ -80,7 +80,7 @@ public interface KeyValueManagement {
      * Get the statuses for all buckets
      * @return list of statuses
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     List<KeyValueStatus> getStatuses() throws IOException, JetStreamApiException;
@@ -89,7 +89,7 @@ public interface KeyValueManagement {
      * Deletes an existing bucket. Will throw a JetStreamApiException if the delete fails.
      * @param bucketName the stream name to use.
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     void delete(String bucketName) throws IOException, JetStreamApiException;

src/main/java/io/nats/client/MessageConsumer.java

@@ -24,7 +24,8 @@
 public interface MessageConsumer extends AutoCloseable {
     /**
      * Gets the consumer name associated with the subscription.
-     * Some simplified consumer types do not support this, so it might be null.
+     * For simplified consumers, this value can be null unless
+     * the consumer info was manually read via {@link #getConsumerInfo()}.
      * @return the consumer name
      */
     String getConsumerName();
@@ -33,21 +34,22 @@ public interface MessageConsumer extends AutoCloseable {
      * Gets information about the consumer behind this subscription.
      * @return consumer information
      * @throws IOException covers various communication issues with the NATS
-     *         server such as timeout or interruption
+     *         server, such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
     ConsumerInfo getConsumerInfo() throws IOException, JetStreamApiException;
 
     /**
      * Gets information about the consumer behind this subscription.
-     * This returns the last read version of Consumer Info, which could technically be out of date.
+     * This returns the last read version of Consumer Info,
+     * which could be null or out of date.
      * @return consumer information
      */
     ConsumerInfo getCachedConsumerInfo();
 
     /**
-     * Use {@link close()} to unsubscribe. Stop will not unsubcribe or clean up resources.
-     * The consumer will finish all pull request already in progress, but will not start any new ones.
+     * Use {@link #close()} to unsubscribe. Stop will not unsubcribe or clean up resources.
+     * The consumer will finish all pull requests already in progress, but will not start any new ones.
      */
     void stop();
 

src/main/java/io/nats/client/SubscribeOptions.java

@@ -34,8 +34,8 @@ public abstract class SubscribeOptions {
     protected final boolean ordered;
     protected final long messageAlarmTime;
     protected final ConsumerConfiguration consumerConfig;
-    protected final long pendingMessageLimit; // Only applicable for non dispatched (sync) push consumers.
-    protected final long pendingByteLimit; // Only applicable for non dispatched (sync) push consumers.
+    protected final long pendingMessageLimit; // Only applicable for non-dispatched (sync) push consumers.
+    protected final long pendingByteLimit; // Only applicable for non-dispatched (sync) push consumers.
     protected final String name;
 
     protected SubscribeOptions(Builder<?, ?> builder, boolean isPull,
@@ -210,15 +210,15 @@ public ConsumerConfiguration getConsumerConfiguration() {
     }
 
     /**
-     * Gets the pending message limit. Only applicable for non dispatched (sync) push consumers.
+     * Gets the pending message limit. Only applicable for non-dispatched (sync) push consumers.
      * @return the message limit
      */
     public long getPendingMessageLimit() {
         return pendingMessageLimit;
     }
 
     /**
-     * Gets the pending byte limit. Only applicable for non dispatched (sync) push consumers.
+     * Gets the pending byte limit. Only applicable for non-dispatched (sync) push consumers.
      * @return the byte limit
      */
     public long getPendingByteLimit() {
@@ -256,7 +256,7 @@ protected static abstract class Builder<B, SO> {
         protected abstract B getThis();
 
         /**
-         * Specify the stream to attach to. If not supplied the stream will be looked up by subject.
+         * Specify the stream to attach to. If not supplied, the stream will look up by the stream by subject.
          * Null or empty clears the field.
          * @param stream the name of the stream
          * @return the builder
@@ -269,8 +269,8 @@ public B stream(String stream) {
         /**
          * Specify binding to an existing consumer via name.
          * The client validates regular (non-fast)
-         * binds to ensure that provided consumer configuration
-         * is consistent with the server version and that
+         * binds to ensure that the provided consumer configuration
+         * is consistent with the server version and that the
          * consumer type (push versus pull) matches the subscription type.
          * @return the builder
          * @param bind whether to bind or not