simplified ordered consumer from context

nats-io · Aug 3, 2023 · ded756a · ded756a
1 parent ce2dba9
commit ded756a
Show file tree

Hide file tree

Showing 22 changed files with 486 additions and 329 deletions.
diff --git a/src/examples/java/io/nats/examples/jetstream/simple/ContextExample.java b/src/examples/java/io/nats/examples/jetstream/simple/ContextExample.java
@@ -62,7 +62,7 @@ public static void main(String[] args) {
             System.out.println("C3. " + consumerContext.getCachedConsumerInfo());
 
             // get a ConsumerContext from the stream context for a pre-existing consumer
-            consumerContext = streamContext.consumerContext(CONSUMER_NAME);
+            consumerContext = streamContext.createConsumerContext(CONSUMER_NAME);
             System.out.println("C4. " + consumerContext.getCachedConsumerInfo());
         }
         catch (JetStreamApiException | IOException | InterruptedException ioe) {

diff --git a/src/examples/java/io/nats/examples/jetstream/simple/IterableConsumerExample.java b/src/examples/java/io/nats/examples/jetstream/simple/IterableConsumerExample.java
@@ -61,7 +61,7 @@ public static void main(String[] args) {
             Thread consumeThread = new Thread(() -> {
                 int count = 0;
                 long start = System.currentTimeMillis();
-                try (IterableConsumer consumer = consumerContext.startIterate()) {
+                try (IterableConsumer consumer = consumerContext.iterate()) {
                     System.out.println("Starting main loop.");
                     while (count < STOP_COUNT) {
                         Message msg = consumer.nextMessage(1000);

diff --git a/src/examples/java/io/nats/examples/jetstream/simple/MessageConsumerExample.java b/src/examples/java/io/nats/examples/jetstream/simple/MessageConsumerExample.java
@@ -53,7 +53,7 @@ public static void main(String[] args) {
             try {
                 streamContext = nc.streamContext(STREAM);
                 streamContext.createOrUpdateConsumer(ConsumerConfiguration.builder().durable(CONSUMER_NAME).build());
-                consumerContext = streamContext.consumerContext(CONSUMER_NAME);
+                consumerContext = streamContext.createConsumerContext(CONSUMER_NAME);
             }
             catch (JetStreamApiException | IOException e) {
                 // JetStreamApiException:
@@ -79,7 +79,7 @@ public static void main(String[] args) {
 
             // create the consumer then use it
             try {
-                MessageConsumer consumer = consumerContext.startConsume(handler);
+                MessageConsumer consumer = consumerContext.consume(handler);
                 latch.await();
                 // once the consumer is stopped, the client will drain messages
                 System.out.println("Stop the consumer...");

diff --git a/src/examples/java/io/nats/examples/jetstream/simple/QueueExample.java b/src/examples/java/io/nats/examples/jetstream/simple/QueueExample.java
@@ -128,7 +128,7 @@ static class HandlerConsumerHolder extends ConsumerHolder {
 
         public HandlerConsumerHolder(int id, StreamContext sc, CountDownLatch latch) throws JetStreamApiException, IOException {
             super(id, sc, latch);
-            messageConsumer = consumerContext.startConsume(msg -> {
+            messageConsumer = consumerContext.consume(msg -> {
                 thisReceived.incrementAndGet();
                 latch.countDown();
                 String data = new String(msg.getData(), StandardCharsets.US_ASCII);
@@ -150,7 +150,7 @@ static class IterableConsumerHolder extends ConsumerHolder {
 
         public IterableConsumerHolder(int id, StreamContext sc, CountDownLatch latch) throws JetStreamApiException, IOException {
             super(id, sc, latch);
-            iterableConsumer = consumerContext.startIterate();
+            iterableConsumer = consumerContext.iterate();
             t = new Thread(() -> {
                 while (latch.getCount() > 0) {
                     try {
@@ -187,7 +187,7 @@ public ConsumerHolder(int id, StreamContext sc, CountDownLatch latch) throws Jet
             this.id = id;
             thisReceived = new AtomicInteger();
             this.latch = latch;
-            consumerContext = sc.consumerContext(CONSUMER_NAME);
+            consumerContext = sc.createConsumerContext(CONSUMER_NAME);
         }
 
         public abstract void stop() throws InterruptedException;

diff --git a/src/main/java/io/nats/client/BaseConsumeOptions.java b/src/main/java/io/nats/client/BaseConsumeOptions.java
@@ -51,7 +51,7 @@ protected BaseConsumeOptions(Builder b) {
         idleHeartbeat = Math.min(MAX_HEARTBEAT_MILLIS, expiresIn * MAX_IDLE_HEARTBEAT_PERCENT / 100);
     }
 
-    public long getExpiresIn() {
+    public long getExpiresInMillis() {
         return expiresIn;
     }
 

diff --git a/src/main/java/io/nats/client/ConsumerContext.java b/src/main/java/io/nats/client/ConsumerContext.java
@@ -121,7 +121,7 @@ public interface ConsumerContext {
      *         server such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
-    IterableConsumer startIterate() throws IOException, JetStreamApiException;
+    IterableConsumer iterate() throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running IterableConsumer with custom ConsumeOptions. See {@link IterableConsumer} and {@link ConsumeOptions}
@@ -132,7 +132,7 @@ public interface ConsumerContext {
      *         server such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
-    IterableConsumer startIterate(ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
+    IterableConsumer iterate(ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running MessageConsumer with default ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
@@ -142,7 +142,7 @@ public interface ConsumerContext {
      *         server such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
-    MessageConsumer startConsume(MessageHandler handler) throws IOException, JetStreamApiException;
+    MessageConsumer consume(MessageHandler handler) throws IOException, JetStreamApiException;
 
     /**
      * Start a long-running MessageConsumer with custom ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
@@ -153,5 +153,5 @@ public interface ConsumerContext {
      *         server such as timeout or interruption
      * @throws JetStreamApiException the request had an error related to the data
      */
-    MessageConsumer startConsume(MessageHandler handler, ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
+    MessageConsumer consume(MessageHandler handler, ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
 }
diff --git a/src/main/java/io/nats/client/MessageConsumer.java b/src/main/java/io/nats/client/MessageConsumer.java
@@ -48,4 +48,8 @@ public interface MessageConsumer extends AutoCloseable {
      * @throws InterruptedException if one is thrown, in order to propagate it up
      */
     void stop(long timeout) throws InterruptedException;
+
+    boolean isStopped();
+
+    boolean isFinished();
 }
diff --git a/src/main/java/io/nats/client/StreamContext.java b/src/main/java/io/nats/client/StreamContext.java
@@ -73,15 +73,25 @@ public interface StreamContext {
     PurgeResponse purge(PurgeOptions options) throws IOException, JetStreamApiException;
 
     /**
-     * Create a consumer context for on the context's stream and specific named consumer.
+     * Create a consumer context for the context's stream and specific named consumer.
      * Verifies that the consumer exists.
      * @param consumerName the name of the consumer
      * @return a ConsumerContext object
      * @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
      */
-    ConsumerContext consumerContext(String consumerName) throws IOException, JetStreamApiException;
+    ConsumerContext createConsumerContext(String consumerName) throws IOException, JetStreamApiException;
+
+    /**
+     * Create an ordered consumer context for the context's stream.
+     * @param config the configuration for the ordered consumer
+     * @return a ConsumerContext object
+     * @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
+     */
+    ConsumerContext createOrderedConsumer(OrderedConsumerConfiguration config) throws IOException, JetStreamApiException;
 
     /**
      * Management function to create or update a consumer on this stream.
@@ -194,50 +204,4 @@ public interface StreamContext {
      * @return true if the delete succeeded
      */
     boolean deleteMessage(long seq, boolean erase) throws IOException, JetStreamApiException;
-
-    /**
-     * Start a long-running Ordered IterableConsumer. See {@link IterableConsumer} and {@link ConsumeOptions}
-     * IterableConsumer requires the developer call nextMessage.
-     * @param config the configuration for the ordered consumer
-     * @return the IterableConsumer instance
-     * @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
-     */
-    IterableConsumer startOrderedIterate(OrderedConsumerConfiguration config) throws IOException, JetStreamApiException;
-
-    /**
-     * Start a long-running Ordered IterableConsumer with custom ConsumeOptions. See {@link IterableConsumer} and {@link ConsumeOptions}
-     * IterableConsumer requires the developer call nextMessage.
-     * @param config the configuration for the ordered consumer
-     * @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
-     * @throws JetStreamApiException the request had an error related to the data
-     */
-    IterableConsumer startOrderedIterate(OrderedConsumerConfiguration config, ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
-
-    /**
-     * Start a long-running Ordered MessageConsumer. See {@link MessageConsumer} and  {@link ConsumeOptions}
-     * @param config the configuration for the ordered consumer
-     * @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 JetStreamApiException the request had an error related to the data
-     */
-    MessageConsumer startOrderedConsume(OrderedConsumerConfiguration config, MessageHandler handler) throws IOException, JetStreamApiException;
-
-    /**
-     * Start a long-running Ordered MessageConsumer with custom ConsumeOptions. See {@link MessageConsumer} and  {@link ConsumeOptions}
-     * @param config the configuration for the ordered consumer
-     * @param handler the MessageHandler used for receiving messages.
-     * @param consumeOptions the custom consume options
-     * @return the MessageConsumer instance
-     * @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 startOrderedConsume(OrderedConsumerConfiguration config, MessageHandler handler, ConsumeOptions consumeOptions) throws IOException, JetStreamApiException;
 }
diff --git a/src/main/java/io/nats/client/impl/NatsConnection.java b/src/main/java/io/nats/client/impl/NatsConnection.java
@@ -2148,15 +2148,15 @@ public StreamContext streamContext(String streamName, JetStreamOptions options)
      */
     @Override
     public ConsumerContext consumerContext(String streamName, String consumerName) throws IOException, JetStreamApiException {
-        return streamContext(streamName).consumerContext(consumerName);
+        return streamContext(streamName).createConsumerContext(consumerName);
     }
 
     /**
      * {@inheritDoc}
      */
     @Override
     public ConsumerContext consumerContext(String streamName, String consumerName, JetStreamOptions options) throws IOException, JetStreamApiException {
-        return streamContext(streamName, options).consumerContext(consumerName);
+        return streamContext(streamName, options).createConsumerContext(consumerName);
     }
 
     /**