Merge pull request #1433 from nats-io/prioritized

(2.12) Prioritized Consumer Support

Author: scottf

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

@@ -45,6 +45,7 @@ public class BaseConsumeOptions implements JsonSerializable {
     protected final int thresholdPercent;
     protected final long idleHeartbeat;
     protected final String group;
+    protected final int priority;
     protected final long minPending;
     protected final long minAckPending;
     protected final boolean raiseStatusWarnings;
@@ -67,6 +68,7 @@ protected BaseConsumeOptions(Builder<?, ?> b) {
         idleHeartbeat = Math.min(MAX_HEARTBEAT_MILLIS, expiresIn * MAX_IDLE_HEARTBEAT_PERCENT / 100);
 
         this.group = b.group;
+        this.priority = b.priority;
         this.minPending = b.minPending;
         this.minAckPending = b.minAckPending;
         raiseStatusWarnings = b.raiseStatusWarnings;
@@ -82,6 +84,7 @@ public String toJson() {
         addField(sb, IDLE_HEARTBEAT, idleHeartbeat);
         addField(sb, THRESHOLD_PERCENT, thresholdPercent);
         addField(sb, GROUP, group);
+        addField(sb, PRIORITY, priority);
         addField(sb, MIN_PENDING, minPending);
         addField(sb, MIN_ACK_PENDING, minAckPending);
         addFldWhenTrue(sb, RAISE_STATUS_WARNINGS, raiseStatusWarnings);
@@ -111,6 +114,10 @@ public String getGroup() {
         return group;
     }
 
+    public int getPriority() {
+        return priority;
+    }
+
     public long getMinPending() {
         return minPending;
     }
@@ -126,6 +133,7 @@ protected static abstract class Builder<B, CO> {
         protected long expiresIn = DEFAULT_EXPIRES_IN_MILLIS;
         protected boolean raiseStatusWarnings = false;
         protected String group;
+        protected int priority;
         protected long minPending = -1;
         protected long minAckPending = -1;
 
@@ -153,6 +161,7 @@ public B jsonValue(JsonValue jsonValue) {
             thresholdPercent(readInteger(jsonValue, THRESHOLD_PERCENT, -1));
             raiseStatusWarnings(readBoolean(jsonValue, RAISE_STATUS_WARNINGS, false));
             group(readStringEmptyAsNull(jsonValue, GROUP));
+            priority(readInteger(jsonValue, PRIORITY, 0));
             minPending(readLong(jsonValue, MIN_PENDING, -1));
             minAckPending(readLong(jsonValue, MIN_ACK_PENDING, -1));
             return getThis();
@@ -239,6 +248,16 @@ public B group(String group) {
             return getThis();
         }
 
+        /**
+         * Sets the priority for the group
+         * @param priority the priority
+         * @return Builder
+         */
+        public B priority(int priority) {
+            this.priority = priority;
+            return getThis();
+        }
+
         /**
          * When specified, the consumer will only receive messages when the consumer has at least this many pending messages.
          * @param minPending the min pending

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

@@ -182,4 +182,15 @@ public interface BaseConsumerContext {
      */
     @NonNull
     MessageConsumer consume(@NonNull ConsumeOptions consumeOptions, @Nullable Dispatcher dispatcher, @NonNull MessageHandler handler) throws IOException, JetStreamApiException;
+
+// TODO - PINNED CONSUMER SUPPORT
+//    /**
+//     * Unpins this consumer
+//     * @param group the group name of the consumer's group
+//     * @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
+//     * @return true if the delete succeeded
+//     */
+//    boolean unpin(String group) throws IOException, JetStreamApiException;
 }

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

@@ -357,6 +357,19 @@ public interface JetStreamManagement {
      */
     boolean deleteMessage(String streamName, long seq, boolean erase) throws IOException, JetStreamApiException;
 
+// TODO - PINNED CONSUMER SUPPORT
+//    /**
+//     * Unpins a consumer
+//     * @param streamName name of the stream
+//     * @param consumerName name of consumer
+//     * @param consumerGroup name of the consumer's group
+//     * @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
+//     * @return true if the delete succeeded
+//     */
+//    boolean unpinConsumer(String streamName, String consumerName, String consumerGroup) throws IOException, JetStreamApiException;
+
     /**
      * Gets a context for publishing and subscribing to subjects backed by Jetstream streams
      * and consumers, using the same connection and JetStreamOptions as the management.

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

@@ -33,6 +33,7 @@ public class PullRequestOptions implements JsonSerializable {
     private final Duration expiresIn;
     private final Duration idleHeartbeat;
     private final String group;
+    private final int priority;
     private final long minPending;
     private final long minAckPending;
 
@@ -43,6 +44,7 @@ public PullRequestOptions(Builder b) {
         this.expiresIn = b.expiresIn;
         this.idleHeartbeat = b.idleHeartbeat;
         this.group = b.group;
+        this.priority = b.priority;
         this.minPending = b.minPending < 0 ? -1 : b.minPending;
         this.minAckPending = b.minAckPending < 0 ? -1 : b.minAckPending;
     }
@@ -56,13 +58,20 @@ public String toJson() {
         JsonUtils.addFldWhenTrue(sb, NO_WAIT, noWait);
         JsonUtils.addFieldAsNanos(sb, EXPIRES, expiresIn);
         JsonUtils.addFieldAsNanos(sb, IDLE_HEARTBEAT, idleHeartbeat);
-
         JsonUtils.addField(sb, GROUP, group);
+        JsonUtils.addFieldWhenGtZero(sb, PRIORITY, priority);
+// TODO - PINNED CONSUMER SUPPORT
+//        JsonUtils.addField(sb, ID, getPinId());
         JsonUtils.addField(sb, MIN_PENDING, minPending);
         JsonUtils.addField(sb, MIN_ACK_PENDING, minAckPending);
         return JsonUtils.endJson(sb).toString();
     }
 
+// TODO - PINNED CONSUMER SUPPORT
+//    protected String getPinId() {
+//        return null;
+//    }
+
     /**
      * Get the batch size option value
      * @return the batch size
@@ -107,6 +116,8 @@ public String getGroup() {
         return group;
     }
 
+    public int getPriority() { return priority; }
+
     public long getMinPending() {
         return minPending;
     }
@@ -140,6 +151,7 @@ public static class Builder {
         private Duration expiresIn;
         private Duration idleHeartbeat;
         private String group;
+        private int priority;
         private long minPending = -1;
         private long minAckPending = -1;
 
@@ -233,6 +245,16 @@ public Builder group(String group) {
             return this;
         }
 
+        /**
+         * Sets the priority within the group. Priority must be between 0 and 9 inclusive.
+         * @param priority the priority
+         * @return Builder
+         */
+        public Builder priority(int priority) {
+            this.priority = priority;
+            return this;
+        }
+
         /**
          * When specified, the pull request will only receive messages when the consumer has at least this many pending messages.
          * @param minPending the min pending
@@ -261,6 +283,9 @@ public Builder minAckPending(long minAckPending) {
          */
         public PullRequestOptions build() {
             validateGtZero(batchSize, "Pull batch size");
+            if (priority < 0 || priority > 9) {
+                throw new IllegalArgumentException("Priority must be between 0 and 9 inclusive.");
+            }
             if (idleHeartbeat != null) {
                 long idleNanosTemp = idleHeartbeat.toNanos() * 2;
                 if (idleNanosTemp > 0) {

src/main/java/io/nats/client/api/ConsumerConfiguration.java

@@ -91,6 +91,7 @@ public class ConsumerConfiguration implements JsonSerializable {
     protected final List<String> filterSubjects;
     protected final List<String> priorityGroups;
     protected final PriorityPolicy priorityPolicy;
+    protected final Duration priorityTimeout;
 
     protected ConsumerConfiguration(ConsumerConfiguration cc) {
         this.deliverPolicy = cc.deliverPolicy;
@@ -124,6 +125,7 @@ protected ConsumerConfiguration(ConsumerConfiguration cc) {
         this.filterSubjects = cc.filterSubjects == null ? null : new ArrayList<>(cc.filterSubjects);
         this.priorityGroups = cc.priorityGroups == null ? null : new ArrayList<>(cc.priorityGroups);
         this.priorityPolicy = cc.priorityPolicy;
+        this.priorityTimeout = cc.priorityTimeout;
     }
 
     // For the builder
@@ -165,6 +167,7 @@ protected ConsumerConfiguration(Builder b)
 
         this.priorityGroups = b.priorityGroups;
         this.priorityPolicy = b.priorityPolicy;
+        this.priorityTimeout = b.priorityTimeout;
     }
 
     /**
@@ -215,6 +218,8 @@ else if (filterSubjects.size() == 1) {
         if (priorityPolicy != null && priorityPolicy != DEFAULT_PRIORITY_POLICY) {
             JsonUtils.addField(sb, PRIORITY_POLICY, priorityPolicy.toString());
         }
+        JsonUtils.addFieldAsNanos(sb, PRIORITY_TIMEOUT, priorityTimeout);
+
         return endJson(sb).toString();
     }
 
@@ -497,14 +502,23 @@ public ZonedDateTime getPauseUntil() {
     }
 
     /**
-     * Gets the priority policy of this consumer configuration.
+     * Gets the priority policy of this consumer configuration. Defaults to PriorityPolicy.None
      * @return the priority policy.
      */
     @NonNull
     public PriorityPolicy getPriorityPolicy() {
         return GetOrDefault(priorityPolicy);
     }
 
+    /**
+     * For pinned_client priority policy how long before the client times out
+     * @return the duration
+     */
+    @Nullable
+    public Duration getPriorityTimeout() {
+        return priorityTimeout;
+    }
+
     /**
      * Gets whether deliver policy of this consumer configuration was set or left unset
      * @return true if the policy was set, false if the policy was not set
@@ -641,6 +655,14 @@ public boolean priorityPolicyWasSet() {
         return priorityPolicy != null;
     }
 
+    /**
+     * Gets whether priority timeout for this consumer configuration was set or left unset
+     * @return true if the timeout was set, false if the timeout was not set
+     */
+    public boolean priorityTimeoutWasSet() {
+        return priorityTimeout != null;
+    }
+
     /**
      * Creates a builder for the options.
      * @return a publish options builder
@@ -703,6 +725,7 @@ public static class Builder {
 
         private List<String> priorityGroups;
         private PriorityPolicy priorityPolicy;
+        private Duration priorityTimeout;
 
         /**
          * Construct the builder
@@ -760,6 +783,7 @@ public Builder(ConsumerConfiguration cc) {
                     this.priorityGroups = new ArrayList<>(cc.priorityGroups);
                 }
                 this.priorityPolicy = cc.priorityPolicy;
+                this.priorityTimeout = cc.priorityTimeout;
             }
         }
 
@@ -843,6 +867,7 @@ public Builder jsonValue(JsonValue jsonValue) {
 
             priorityGroups(readOptionalStringList(jsonValue, PRIORITY_GROUPS));
             priorityPolicy(PriorityPolicy.get(readString(jsonValue, PRIORITY_POLICY)));
+            priorityTimeout(readNanos(jsonValue, PRIORITY_TIMEOUT));
 
             return this;
         }
@@ -1405,6 +1430,26 @@ public Builder priorityPolicy(PriorityPolicy policy) {
             return this;
         }
 
+        /**
+         * Sets the priority policy timeout
+         * @param priorityTimeout the timeout
+         * @return Builder
+         */
+        public Builder priorityTimeout(Duration priorityTimeout) {
+            this.priorityTimeout = normalize(priorityTimeout);
+            return this;
+        }
+
+        /**
+         * Sets the priority policy timeout
+         * @param priorityTimeoutMillis the timeout in milliseconds
+         * @return Builder
+         */
+        public Builder priorityTimeout(long priorityTimeoutMillis) {
+            this.priorityTimeout = normalizeDuration(priorityTimeoutMillis);
+            return this;
+        }
+
         /**
          * Builds the ConsumerConfiguration
          * @return The consumer configuration.

src/main/java/io/nats/client/api/ConsumerInfo.java

@@ -21,6 +21,7 @@
 
 import java.time.Duration;
 import java.time.ZonedDateTime;
+import java.util.List;
 
 import static io.nats.client.support.ApiConstants.*;
 import static io.nats.client.support.JsonValueUtils.*;
@@ -46,6 +47,7 @@ public class ConsumerInfo extends ApiResponse<ConsumerInfo> {
     private final ClusterInfo clusterInfo;
     private final boolean pushBound;
     private final ZonedDateTime timestamp;
+    private final List<PriorityGroupState> priorityGroupStates;
 
     public ConsumerInfo(Message msg) {
         this(parseMessage(msg));
@@ -69,6 +71,7 @@ public ConsumerInfo(JsonValue vConsumerInfo) {
             clusterInfo = null;
             pushBound = false;
             timestamp = null;
+            priorityGroupStates = null;
         }
         else {
             JsonValue jvConfig = nullValueIsError(jv, CONFIG, JsonValue.EMPTY_MAP) ;
@@ -92,6 +95,8 @@ public ConsumerInfo(JsonValue vConsumerInfo) {
             pushBound = readBoolean(jv, PUSH_BOUND);
 
             timestamp = readDate(jv, TIMESTAMP);
+
+            priorityGroupStates = PriorityGroupState.optionalListOf(readObject(jv, PRIORITY_GROUPS));
         }
     }
 
@@ -224,6 +229,15 @@ public ZonedDateTime getTimestamp() {
         return timestamp;
     }
 
+    /**
+     * The state of Priority Groups
+     * @return the list of Priority Groups, may be null
+     */
+    @Nullable
+    public List<PriorityGroupState> getPriorityGroupStates() {
+        return priorityGroupStates;
+    }
+
     /**
      * A way to more accurately calculate pending during the initial state
      * of the consumer when messages may be unaccounted for in flight