feat: add createdTime(publishTime) into ReceivedRecord (#159)

Author: s12f

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

@@ -1,12 +1,16 @@
 package io.hstream;
 
+import java.time.Instant;
+
 public class ReceivedRecord {
   String recordId;
   Record record;
+  Instant createdTime;
 
-  public ReceivedRecord(String recordId, Record record) {
+  public ReceivedRecord(String recordId, Record record, Instant createdTime) {
     this.recordId = recordId;
     this.record = record;
+    this.createdTime = createdTime;
   }
 
   public String getRecordId() {
@@ -24,4 +28,8 @@ public Record getRecord() {
   public void setRecord(Record record) {
     this.record = record;
   }
+
+  public Instant getCreatedTime() {
+    return createdTime;
+  }
 }

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

@@ -16,6 +16,7 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory
+import java.time.Instant
 import java.util.concurrent.CompletableFuture
 
 class ReaderKtImpl(
@@ -55,14 +56,16 @@ class ReaderKtImpl(
                     RecordUtils.decompress(it).map { receivedHStreamRecord ->
                         val hStreamRecord = receivedHStreamRecord.record
                         val header = RecordUtils.parseRecordHeaderFromHStreamRecord(hStreamRecord)
+                        val publishTime = it.record.publishTime
+                        val createdTime = Instant.ofEpochSecond(publishTime.seconds, publishTime.nanos.toLong())
                         val record = if (RecordUtils.isRawRecord(hStreamRecord)) {
                             val rawRecord = RecordUtils.parseRawRecordFromHStreamRecord(hStreamRecord)
                             Record.newBuilder().rawRecord(rawRecord).partitionKey(header.partitionKey).build()
                         } else {
                             val hRecord = RecordUtils.parseHRecordFromHStreamRecord(hStreamRecord)
                             Record.newBuilder().hRecord(hRecord).partitionKey(header.partitionKey).build()
                         }
-                        ReceivedRecord(GrpcUtils.recordIdFromGrpc(receivedHStreamRecord.recordId), record)
+                        ReceivedRecord(GrpcUtils.recordIdFromGrpc(receivedHStreamRecord.recordId), record, createdTime)
                     }
                 }
                 readFuture.complete(res as MutableList<ReceivedRecord>?)