feat: add createdTime in Stream and Subscription (#157)

Author: s12f

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

@@ -3,19 +3,27 @@
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
+import java.time.Instant;
 import java.util.Objects;
 
 public class Stream {
   private String streamName;
   private int replicationFactor;
   private int backlogDuration;
   private int shardCount;
-
-  public Stream(String streamName, int replicationFactor, int backlogDuration, int shardCount) {
+  private final Instant createdTime;
+
+  Stream(
+      String streamName,
+      int replicationFactor,
+      int backlogDuration,
+      int shardCount,
+      Instant createdTime) {
     this.streamName = streamName;
     this.replicationFactor = replicationFactor;
     this.backlogDuration = backlogDuration;
     this.shardCount = shardCount;
+    this.createdTime = createdTime;
   }
 
   public String getStreamName() {
@@ -50,6 +58,10 @@ public void setShardCount(int shardCount) {
     this.shardCount = shardCount;
   }
 
+  public Instant getCreatedTime() {
+    return createdTime;
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) return true;
@@ -71,6 +83,7 @@ public static final class Builder {
     private int replicationFactor = 1;
     private int backlogDuration = 3600 * 24;
     private int shardCount = 1;
+    private Instant createdTime;
 
     /**
      * @param streamName required, the name of the stream
@@ -109,11 +122,16 @@ public Builder shardCount(int shardCount) {
       return this;
     }
 
+    public Builder createdTime(Instant createdTime) {
+      this.createdTime = createdTime;
+      return this;
+    }
+
     public Stream build() {
       checkNotNull(streamName);
       checkArgument(replicationFactor >= 1 && replicationFactor <= 15);
       checkArgument(shardCount >= 1);
-      return new Stream(streamName, replicationFactor, backlogDuration, shardCount);
+      return new Stream(streamName, replicationFactor, backlogDuration, shardCount, createdTime);
     }
   }
 

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

@@ -2,6 +2,7 @@
 
 import static com.google.common.base.Preconditions.*;
 
+import java.time.Instant;
 import java.util.Objects;
 
 public class Subscription {
@@ -14,6 +15,12 @@ public class Subscription {
 
   private SubscriptionOffset offset;
 
+  public Instant getCreatedTime() {
+    return createdTime;
+  }
+
+  private final Instant createdTime;
+
   public enum SubscriptionOffset {
     EARLIEST,
     LATEST,
@@ -24,12 +31,14 @@ private Subscription(
       String streamName,
       int ackTimeoutSeconds,
       int maxUnackedRecords,
-      SubscriptionOffset offset) {
+      SubscriptionOffset offset,
+      Instant createdTime) {
     this.subscriptionId = subscriptionId;
     this.streamName = streamName;
     this.ackTimeoutSeconds = ackTimeoutSeconds;
     this.maxUnackedRecords = maxUnackedRecords;
     this.offset = offset;
+    this.createdTime = createdTime;
   }
 
   /** @return {@link Subscription.Builder} */
@@ -81,6 +90,8 @@ public static class Builder {
     private int maxUnackedRecords = 10000;
     private SubscriptionOffset offset = SubscriptionOffset.LATEST;
 
+    private Instant createdTime;
+
     public Builder subscription(String subscriptionId) {
       this.subscriptionId = subscriptionId;
       return this;
@@ -106,13 +117,18 @@ public Builder offset(SubscriptionOffset offset) {
       return this;
     }
 
+    public Builder createdTime(Instant createdTime) {
+      this.createdTime = createdTime;
+      return this;
+    }
+
     public Subscription build() {
       checkNotNull(subscriptionId);
       checkNotNull(streamName);
       checkState(ackTimeoutSeconds > 0 && ackTimeoutSeconds < 36000);
       checkState(maxUnackedRecords > 0);
       return new Subscription(
-          subscriptionId, streamName, ackTimeoutSeconds, maxUnackedRecords, offset);
+          subscriptionId, streamName, ackTimeoutSeconds, maxUnackedRecords, offset, createdTime);
     }
   }
 }

client/src/main/java/io/hstream/TaskStatus.java

@@ -5,6 +5,5 @@ public enum TaskStatus {
   TASK_CREATED,
   TASK_RUNNING,
   TASK_CREATION_ABORT,
-  TASK_CONNECTION_ABORT,
   TASK_TERMINATED,
 }

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

@@ -6,6 +6,7 @@
 import io.hstream.internal.RecordId;
 import io.hstream.internal.SpecialOffset;
 import io.hstream.internal.TaskStatusPB;
+import java.time.Instant;
 
 /**
  * A class of utility functions to convert between the GRPC generated classes and the custom classes
@@ -61,10 +62,12 @@ public static io.hstream.internal.Subscription subscriptionToGrpc(Subscription s
   }
 
   public static Subscription subscriptionFromGrpc(io.hstream.internal.Subscription subscription) {
+    var createdTime = subscription.getCreationTime();
     return Subscription.newBuilder().subscription(subscription.getSubscriptionId()).stream(
             subscription.getStreamName())
         .ackTimeoutSeconds(subscription.getAckTimeoutSeconds())
         .offset(subscriptionOffsetFromGrpc(subscription.getOffset()))
+        .createdTime(Instant.ofEpochSecond(createdTime.getSeconds(), createdTime.getNanos()))
         .build();
   }
 
@@ -78,11 +81,14 @@ public static io.hstream.internal.Stream streamToGrpc(Stream stream) {
   }
 
   public static Stream streamFromGrpc(io.hstream.internal.Stream stream) {
-    return new Stream(
-        stream.getStreamName(),
-        stream.getReplicationFactor(),
-        stream.getBacklogDuration(),
-        stream.getShardCount());
+    var createdTime = stream.getCreationTime();
+    return Stream.newBuilder()
+        .streamName(stream.getStreamName())
+        .replicationFactor(stream.getReplicationFactor())
+        .backlogDuration(stream.getBacklogDuration())
+        .shardCount(stream.getShardCount())
+        .createdTime(Instant.ofEpochSecond(createdTime.getSeconds(), createdTime.getNanos()))
+        .build();
   }
 
   public static StreamShardOffset streamShardOffsetFromGrpc(
@@ -183,8 +189,6 @@ public static TaskStatus taskStatusFromInternal(TaskStatusPB statusPB) {
         return TaskStatus.TASK_CREATED;
       case TASK_RUNNING:
         return TaskStatus.TASK_RUNNING;
-      case TASK_CONNECTION_ABORT:
-        return TaskStatus.TASK_CONNECTION_ABORT;
       case TASK_CREATION_ABORT:
         return TaskStatus.TASK_CREATION_ABORT;
       case TASK_TERMINATED:

client/src/main/kotlin/io/hstream/impl/HStreamClientKtImpl.kt

@@ -121,12 +121,12 @@ class HStreamClientKtImpl(bootstrapServerUrls: List<String>, credentials: Channe
         unaryCallBlocked {
             it.createStream(
                 GrpcUtils.streamToGrpc(
-                    Stream(
-                        stream,
-                        replicationFactor.toInt(),
-                        backlogDuration,
-                        shardCnt
-                    )
+                    Stream.newBuilder()
+                        .streamName(stream)
+                        .replicationFactor(replicationFactor.toInt())
+                        .shardCount(shardCnt)
+                        .backlogDuration(backlogDuration)
+                        .build()
                 )
             )
         }