Jetstream Usage (#1193)

Author: roeschter

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

@@ -13,15 +13,101 @@
 
 package io.nats.client;
 
+import io.nats.client.api.ConsumerConfiguration;
 import io.nats.client.api.PublishAck;
+import io.nats.client.api.StorageType;
+import io.nats.client.api.StreamConfiguration;
 import io.nats.client.impl.Headers;
 
 import java.io.IOException;
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 
 /**
- * JetStream context for creation and access to streams and consumers in NATS.
+ * JetStream context for access to streams and consumers in NATS.
+ * 
+ * <h3>Basic usage</h3>
+ * 
+ * <p>{@link #publish(String, byte[]) JetStream.Publish} will send a message on the specified subject, waiting for acknowledgement. 
+ * A  <b>503 No responders</b> error will be received if no stream is listening on said subject.
+ * 
+ * <p>{@link #publishAsync(String, byte[]) PublishAsync} will not wait for acknowledgement but return a {@link CompletableFuture CompletableFuture}, 
+ * which can be checked for acknowledgement at a later point.
+ * 
+ * <p> Use {@link #getStreamContext(String ) getStreamContext(String)} to access a simplified API for <b>consuming/subscribing</b> messages from Jetstream. 
+ * It is <b>recommened</b> to manage consumers explicitely through {@link StreamContext StreamContext} or {@link JetStreamManagement JetStreamManagement}
+ * 
+ * <p>{@link #subscribe(String)} is a convenience method for implicitly creating a consumer on a stream and receiving messages. This method should be used for ephemeral (not durable) conusmers. 
+ * It can create a named durable consumers though Options, but we prefer to avoid creating durable consumers implictly. 
+ * It is <b>recommened</b> to manage consumers explicitely through {@link StreamContext StreamContext} or {@link JetStreamManagement JetStreamManagement}
+ *  
+ *  {@link ConsumerContext ConsumerContext} based subscription. 
+ * 
+ * <h3>Recommended usage for creating streams, consumers, publish and listen on a stream</h3>
+ * <pre>
+ *  io.nats.client.Connection nc = Nats.connect();
+ *	
+ *  //Setting up a stream and a consumer	
+ *  JetStreamManagement jsm = nc.jetStreamManagement();
+ *  StreamConfiguration sc = StreamConfiguration.builder()
+ *    .name("my-stream")
+ *    .storageType(StorageType.File)
+ *    .subjects("foo.*", "bar.*")
+ *    .build();
+ *      
+ *  jsm.addStream(sc);
+ *       
+ *  ConsumerConfiguration consumerConfig = ConsumerConfiguration.builder()
+ *    .durable("my-consumer")
+ *    .build();
+ *		
+ *  jsm.createConsumer("my-stream", consumerConfig);
+ *    	
+ *  //Listening and publishing  
+ *  io.nats.client.JetStream js = nc.jetStream();
+ *  ConsumerContext consumerContext = js.getConsumerContext("my-stream", "my-consumer"); 
+ *  MessageConsumer mc = consumerContext.consume(
+ *    msg -&gt; {
+ *      System.out.println("   Received " + msg.getSubject());
+ *      msg.ack();
+ *    });
+ *    	
+ *  js.publish("foo.joe", "Hello World".getBytes());
+ *  
+ *  //Wait a moment, then stop the MessageConsumer
+ *  Thread.sleep(3000);
+ *  mc.stop();
+ *  
+ *  </pre>
+ *  
+ *	<h3>Recommended usage of asynchronous publishing</h3>
+ *  
+ *  Jetstream messages can be published asynchronously, returning a CompletableFuture. 
+ *  Note that you need to check the Future eventually otherwise the delivery guarantee is the same a regular {@link Connection#publish(String, byte[]) Connection.Publish} 
+ *  
+ *  <p>We are publishing a batch of 100 messages and check for completion afterwards.
+ * 
+ *  <pre>
+ *  int COUNT = 100;
+ *  java.util.concurrent.CompletableFuture&lt;?&gt;[] acks = new java.util.concurrent.CompletableFuture&lt;?&gt;[COUNT];
+ * 
+ *  for( int i=0; i&lt;COUNT; i++ ) {
+ *    acks[i] = js.publishAsync("foo.joe", ("Hello "+i).getBytes());
+ *  }
+ *    	
+ *  //Acknowledgments may arrive out of sequence, but CompletableFuture is handling this for us.
+ *  for( int i=0; i&lt;COUNT; i++ ) {
+ *    try {
+ *      acks[i].get();
+ *    } catch ( Exception e ) {
+ *      //Retry or handle error
+ *    }
+ *  }
+ *  
+ *  //Now we may send anther batch
+ * 
+ * </pre>
+ * 
  */
 public interface JetStream {
 
@@ -458,7 +544,8 @@ public interface JetStream {
     JetStreamSubscription subscribe(String subscribeSubject, PullSubscribeOptions options) throws IOException, JetStreamApiException;
 
     /**
-     * Create an asynchronous subscription to the specified subject in the mode of pull, with additional options
+     * Create an asynchronous subscription to the specified subject in the mode of pull, with additional options.
+     * 
      * @param subscribeSubject The subject to subscribe to
      *                         Can be null or empty when the options have a ConsumerConfiguration that supplies a filter subject.
      * @param dispatcher The dispatcher to handle this subscription
@@ -473,6 +560,25 @@ public interface JetStream {
 
     /**
      * Get a stream context for a specific named stream. Verifies that the stream exists.
+     * 
+     * <p><b>Recommended usage:</b> {@link StreamContext StreamContext} and {@link ConsumerContext ConsumerContext} are the preferred way to interact with existing streams and consume from streams. 
+     * {@link JetStreamManagement JetStreamManagement} should be used to create streams and consumers. {@link ConsumerContext#consume ConsumerContext.consume()} supports both push and pull consumers transparently.
+     * 
+     * <pre>
+     *  nc = Nats.connect();
+     *  Jetstream js = nc.jetStream();
+	 *  StreamContext streamContext = js.getStreamContext("my-stream");
+	 *  ConsumerContext consumerContext = streamContext.getConsumerContext("my-consumer");
+	 *  // Or 
+	 *  // ConsumerContext consumerContext = js.getConsumerContext("my-stream", "my-consumer"); 
+	 *  consumerContext.consume(
+	 *    msg -&gt; {
+	 *      System.out.println("   Received " + msg.getSubject());
+	 *      msg.ack();
+	 *    });
+	 *  </pre>        
+     * 
+     * 
      * @param streamName the name of the stream
      * @return a StreamContext object
      * @throws IOException covers various communication issues with the NATS
@@ -483,6 +589,9 @@ public interface JetStream {
 
     /**
      * Get a consumer context for a specific named stream and specific named consumer.
+     * 
+     * <p><b>Recommended usage:</b> See {@link #getStreamContext(String) getStreamContext(String)} 
+     * 
      * Verifies that the stream and consumer exist.
      * @param streamName the name of the stream
      * @param consumerName the name of the consumer