feat: add timestamp offset (#198)

Author: daleiz

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

@@ -7,13 +7,16 @@ public enum SpecialOffset {
     LATEST;
   }
 
-  private enum OffsetType {
+  public enum OffsetType {
     SPECIAL,
+    TIMESTAMP,
     NORMAL;
   }
 
   private SpecialOffset specialOffset;
   private String recordId;
+  private long timestamp;
+
   private OffsetType offsetType;
 
   public StreamShardOffset(SpecialOffset specialOffset) {
@@ -26,6 +29,11 @@ public StreamShardOffset(String recordId) {
     this.offsetType = OffsetType.NORMAL;
   }
 
+  public StreamShardOffset(long timestamp) {
+    this.timestamp = timestamp;
+    this.offsetType = OffsetType.TIMESTAMP;
+  }
+
   public boolean isSpecialOffset() {
     return offsetType.equals(OffsetType.SPECIAL);
   }
@@ -34,6 +42,14 @@ public boolean isNormalOffset() {
     return offsetType.equals(OffsetType.NORMAL);
   }
 
+  public boolean isTimestampOffset() {
+    return offsetType.equals(OffsetType.TIMESTAMP);
+  }
+
+  public OffsetType getOffsetType() {
+    return offsetType;
+  }
+
   public SpecialOffset getSpecialOffset() {
     if (isSpecialOffset()) {
       return specialOffset;
@@ -49,4 +65,12 @@ public String getNormalOffset() {
       throw new IllegalStateException("subscriptionOffset is not normal offset");
     }
   }
+
+  public long getTimestampOffset() {
+    if (isTimestampOffset()) {
+      return timestamp;
+    } else {
+      throw new IllegalStateException("subscriptionOffset is not timestamp offset");
+    }
+  }
 }

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

@@ -7,8 +7,10 @@
 import com.google.protobuf.util.JsonFormat;
 import io.hstream.*;
 import io.hstream.internal.RecordId;
+import io.hstream.internal.ShardOffset;
 import io.hstream.internal.SpecialOffset;
 import io.hstream.internal.TaskStatusPB;
+import io.hstream.internal.TimestampOffset;
 import java.time.Instant;
 import java.util.stream.Collectors;
 
@@ -96,24 +98,6 @@ public static Stream streamFromGrpc(io.hstream.internal.Stream stream) {
         .build();
   }
 
-  public static StreamShardOffset streamShardOffsetFromGrpc(
-      io.hstream.internal.ShardOffset shardOffset) {
-    if (shardOffset.hasSpecialOffset()) {
-      switch (shardOffset.getSpecialOffset()) {
-        case EARLIEST:
-          return new StreamShardOffset(StreamShardOffset.SpecialOffset.EARLIEST);
-        case LATEST:
-          return new StreamShardOffset(StreamShardOffset.SpecialOffset.LATEST);
-        default:
-          throw new IllegalArgumentException("Unknown ShardOffset : " + shardOffset);
-      }
-    } else if (shardOffset.hasRecordOffset()) {
-      return new StreamShardOffset(recordIdFromGrpc(shardOffset.getRecordOffset()));
-    } else {
-      throw new IllegalArgumentException("Unknown ShardOffset : " + shardOffset);
-    }
-  }
-
   public static io.hstream.internal.ShardOffset streamShardOffsetToGrpc(
       StreamShardOffset shardOffset) {
     if (shardOffset.isSpecialOffset()) {
@@ -133,6 +117,14 @@ public static io.hstream.internal.ShardOffset streamShardOffsetToGrpc(
       return io.hstream.internal.ShardOffset.newBuilder()
           .setRecordOffset(recordIdToGrpc(shardOffset.getNormalOffset()))
           .build();
+    } else if (shardOffset.isTimestampOffset()) {
+      return ShardOffset.newBuilder()
+          .setTimestampOffset(
+              TimestampOffset.newBuilder()
+                  .setTimestampInMs(shardOffset.getTimestampOffset())
+                  .setStrictAccuracy(true)
+                  .build())
+          .build();
     } else {
       throw new IllegalArgumentException("Unknown streamShardOffset : " + shardOffset);
     }