encapsulate the Stream class & documentation (#22)

* encapsulate class Stream

* updated examples in README

* added some javadoc

Author: Yu-zh

README.md

@@ -137,31 +137,27 @@ batchedProducer.flush()
 ### Consume Data from a Stream 
 
 ```java
-
 // first, create a subscription for the stream
 Subscription subscription =
-        Subscription.newBuilder()
-            .setSubscriptionId("my_subscription")
-            .setStreamName("test_stream")
-            .setOffset(
-                SubscriptionOffset.newBuilder()
-                    .setSpecialOffset(SubscriptionOffset.SpecialOffset.LATEST)
-                    .build())
-            .build();
-client.createSubscription(subscription);
-
-// second, create a consumer attacth to the subscription 
+    new Subscription(
+        "my_subscription",
+        "test_stream",
+        new SubscriptionOffset(SubscriptionOffset.SpecialOffset.LATEST));
+    client.createSubscription(subscription);
+
+// second, create a consumer attach to the subscription
 Consumer consumer =
-      client
-          .newConsumer()
-          .subscription("my_subscription")
-          .rawRecordReceiver(
-              (receivedRawRecord, responder) -> {
-                  System.out.println(receivedRawRecord.getRecordId());
-                  responder.ack();})
-          .build();
+    client
+        .newConsumer()
+        .subscription("my_subscription")
+        .rawRecordReceiver(
+            ((receivedRawRecord, responder) -> {
+                System.out.println(receivedRawRecord.getRecordId());
+                responder.ack();
+            }))
+        .build();
 
-// third, start the consumer 
+// third, start the consumer
 consumer.startAsync().awaitRunning();
 
 ```

src/main/java/io/hstream/HStreamClient.java

@@ -1,6 +1,5 @@
 package io.hstream;
 
-import io.hstream.internal.Stream;
 import java.util.List;
 
 /** HstreamClient implement the hstream client, user can use it to interact with server */

src/main/java/io/hstream/RecordId.java

@@ -2,28 +2,41 @@
 
 import java.util.Objects;
 
+/** A class for storing identification information of data records */
 public class RecordId {
 
+  /** An identification number for the data record that is unique among all the data records */
   private long batchId;
+  /** An index number identifies the data record in its own batch */
   private int batchIndex;
 
+  /**
+   * A constructor for RecordId
+   *
+   * @param batchId the unique identification number of the data record in the database
+   * @param batchIndex the index of the data record in its own batch
+   */
   public RecordId(long batchId, int batchIndex) {
     this.batchId = batchId;
     this.batchIndex = batchIndex;
   }
 
+  /** get the unique identification number of the data record */
   public long getBatchId() {
     return batchId;
   }
 
+  /** get the index of the data record in its own batch */
   public int getBatchIndex() {
     return batchIndex;
   }
 
+  /** update the identification number */
   public void setBatchId(long batchId) {
     this.batchId = batchId;
   }
 
+  /** update the data record index */
   public void setBatchIndex(int batchIndex) {
     this.batchIndex = batchIndex;
   }

src/main/java/io/hstream/Stream.java

@@ -0,0 +1,58 @@
+package io.hstream;
+
+import java.util.Objects;
+
+/** A class for storing information about streams */
+public class Stream {
+  private String streamName;
+  private int replicationFactor;
+
+  /**
+   * a constructor for a real-time data stream
+   *
+   * @param streamName the name of the stream
+   * @param replicationFactor the number of replicas to be stored in the database
+   */
+  public Stream(String streamName, int replicationFactor) {
+    this.streamName = streamName;
+    this.replicationFactor = replicationFactor;
+  }
+
+  /** get the name of the stream */
+  public String getStreamName() {
+    return streamName;
+  }
+
+  /** get the replication factor of the stream */
+  public int getReplicationFactor() {
+    return replicationFactor;
+  }
+
+  /** update the name of the stream */
+  public void setStreamName(String streamName) {
+    this.streamName = streamName;
+  }
+
+  /** update the replication factor of the stream */
+  public void setReplicationFactor(int replicationFactor) {
+    this.replicationFactor = replicationFactor;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    Stream stream = (Stream) o;
+    return replicationFactor == stream.replicationFactor
+        && Objects.equals(streamName, stream.streamName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(streamName, replicationFactor);
+  }
+}

src/main/java/io/hstream/Subscription.java

@@ -2,39 +2,57 @@
 
 import java.util.Objects;
 
+/** A class for storing information about subscriptions */
 public class Subscription {
 
+  /** An identifier for the subscription */
   private String subscriptionId;
+  /** The name of the stream being subscribed to */
   private String streamName;
+  /** The offset that indicates the position to start consuming data from the stream */
   private SubscriptionOffset subscriptionOffset;
 
+  /**
+   * A constructor for subscriptions
+   *
+   * @param subscriptionId An identifier for the subscription
+   * @param streamName The name of the stream being subscribed to
+   * @param subscriptionOffset A {@link SubscriptionOffset} to indicate the position to start
+   *     consuming data
+   */
   public Subscription(
       String subscriptionId, String streamName, SubscriptionOffset subscriptionOffset) {
     this.subscriptionId = subscriptionId;
     this.streamName = streamName;
     this.subscriptionOffset = subscriptionOffset;
   }
 
+  /** get the identifier of the subscription */
   public String getSubscriptionId() {
     return subscriptionId;
   }
 
+  /** get the name of stream being subscribed to */
   public String getStreamName() {
     return streamName;
   }
 
+  /** get the subscription offset */
   public SubscriptionOffset getSubscriptionOffset() {
     return subscriptionOffset;
   }
 
+  /** update the identifier of the subscription */
   public void setSubscriptionId(String subscriptionId) {
     this.subscriptionId = subscriptionId;
   }
 
+  /** update the name of the stream */
   public void setStreamName(String streamName) {
     this.streamName = streamName;
   }
 
+  /** update the subscription offset */
   public void setSubscriptionOffset(SubscriptionOffset subscriptionOffset) {
     this.subscriptionOffset = subscriptionOffset;
   }

src/main/java/io/hstream/SubscriptionOffset.java

@@ -2,23 +2,36 @@
 
 import java.util.Objects;
 
+/** A class with information about subscription offset */
 public class SubscriptionOffset {
+  /** Subscription offsets that can be used */
   public enum SpecialOffset {
+    /** Start consuming messages from the earliest position in the stream */
     EARLIEST,
+    /**
+     * Ignore the history messages up to the moment of subscription and ead the messages thereafter.
+     */
     LATEST,
     UNRECOGNIZED;
   }
 
   private SpecialOffset specialOffset;
 
+  /**
+   * A constructor for subscription offset
+   *
+   * @param specialOffset One of the offsets defined in {@link SpecialOffset}
+   */
   public SubscriptionOffset(SpecialOffset specialOffset) {
     this.specialOffset = specialOffset;
   }
 
+  /** get the offset */
   public SpecialOffset getSpecialOffset() {
     return specialOffset;
   }
 
+  /** update the offset */
   public void setSpecialOffset(SpecialOffset specialOffset) {
     this.specialOffset = specialOffset;
   }

src/main/java/io/hstream/impl/ClientImpl.java

@@ -8,7 +8,6 @@
 import io.hstream.internal.DeleteSubscriptionRequest;
 import io.hstream.internal.HStreamApiGrpc;
 import io.hstream.internal.ListStreamsResponse;
-import io.hstream.internal.Stream;
 import io.hstream.util.GrpcUtils;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -48,9 +47,9 @@ public QueryerBuilder newQueryer() {
 
   @Override
   public void createStream(String streamName) {
-    Stream stream = Stream.newBuilder().setStreamName(streamName).setReplicationFactor(3).build();
+    Stream stream = new Stream(streamName, 3);
 
-    blockingStub.createStream(stream);
+    blockingStub.createStream(GrpcUtils.streamToGrpc(stream));
   }
 
   @Override
@@ -65,7 +64,9 @@ public void deleteStream(String streamName) {
   public List<Stream> listStreams() {
     Empty empty = Empty.newBuilder().build();
     ListStreamsResponse listStreamsResponse = blockingStub.listStreams(empty);
-    return listStreamsResponse.getStreamsList();
+    return listStreamsResponse.getStreamsList().stream()
+        .map(GrpcUtils::streamFromGrpc)
+        .collect(Collectors.toList());
   }
 
   @Override

src/main/java/io/hstream/util/GrpcUtils.java

@@ -2,6 +2,10 @@
 
 import io.hstream.*;
 
+/**
+ * A class of utility functions to convert between the GRPC generated classes and the custom classes
+ * e.g. {@link RecordId}, {@link Subscription}, and {@link Stream}
+ */
 public class GrpcUtils {
 
   public static io.hstream.internal.RecordId recordIdToGrpc(RecordId recordId) {
@@ -69,4 +73,15 @@ public static Subscription subscriptionFromGrpc(io.hstream.internal.Subscription
         subscription.getStreamName(),
         subscriptionOffsetFromGrpc(subscription.getOffset()));
   }
+
+  public static io.hstream.internal.Stream streamToGrpc(Stream stream) {
+    return io.hstream.internal.Stream.newBuilder()
+        .setStreamName(stream.getStreamName())
+        .setReplicationFactor(stream.getReplicationFactor())
+        .build();
+  }
+
+  public static Stream streamFromGrpc(io.hstream.internal.Stream stream) {
+    return new Stream(stream.getStreamName(), stream.getReplicationFactor());
+  }
 }