review comments

Signed-off-by: Yashash H L <[email protected]>

pkg/mapper/interface.go

@@ -7,9 +7,13 @@ import (
 
 // Datum contains methods to get the payload information.
 type Datum interface {
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }
 

pkg/mapstreamer/interface.go

@@ -7,9 +7,13 @@ import (
 
 // Datum contains methods to get the payload information.
 type Datum interface {
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }
 

pkg/reducer/interface.go

@@ -7,9 +7,13 @@ import (
 
 // Datum contains methods to get the payload information.
 type Datum interface {
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }
 

pkg/reducer/task_manager.go

@@ -166,5 +166,10 @@ func generateKey(window *v1.Window, keys []string) string {
 }
 
 func buildDatum(request *v1.ReduceRequest) Datum {
-	return NewHandlerDatum(request.Payload.GetValue(), request.Payload.EventTime.AsTime(), request.Payload.Watermark.AsTime(), request.Payload.Headers)
+	return NewHandlerDatum(
+		request.GetPayload().GetValue(),
+		request.GetPayload().GetEventTime().AsTime(),
+		request.GetPayload().GetWatermark().AsTime(),
+		request.GetPayload().GetHeaders(),
+	)
 }

pkg/reducestreamer/interface.go

@@ -7,9 +7,13 @@ import (
 
 // Datum contains methods to get the payload information.
 type Datum interface {
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }
 

pkg/reducestreamer/task_manager.go

@@ -173,5 +173,10 @@ func generateKey(window *v1.Window, keys []string) string {
 }
 
 func buildDatum(request *v1.ReduceRequest) Datum {
-	return NewHandlerDatum(request.Payload.GetValue(), request.Payload.EventTime.AsTime(), request.Payload.Watermark.AsTime(), request.Payload.Headers)
+	return NewHandlerDatum(
+		request.GetPayload().GetValue(),
+		request.GetPayload().GetEventTime().AsTime(),
+		request.GetPayload().GetWatermark().AsTime(),
+		request.GetPayload().GetHeaders(),
+	)
 }

pkg/sessionreducer/interface.go

@@ -7,9 +7,13 @@ import (
 
 // Datum contains methods to get the payload information.
 type Datum interface {
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }
 

pkg/sessionreducer/task_manager.go

@@ -320,5 +320,10 @@ func generateKey(keyedWindow *v1.KeyedWindow) string {
 }
 
 func buildDatum(payload *v1.SessionReduceRequest_Payload) Datum {
-	return NewHandlerDatum(payload.GetValue(), payload.EventTime.AsTime(), payload.Watermark.AsTime(), payload.Headers)
+	return NewHandlerDatum(
+		payload.GetValue(),
+		payload.GetEventTime().AsTime(),
+		payload.GetWatermark().AsTime(),
+		payload.GetHeaders(),
+	)
 }

pkg/sinker/interface.go

@@ -7,11 +7,17 @@ import (
 
 // Datum is the interface of incoming message payload for sink function.
 type Datum interface {
+	// Keys returns the keys of the message.
 	Keys() []string
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// ID returns the ID of the message.
 	ID() string
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }
 

pkg/sourcetransformer/interface.go

@@ -7,9 +7,13 @@ import (
 
 // Datum contains methods to get the payload information.
 type Datum interface {
+	// Value returns the payload of the message.
 	Value() []byte
+	// EventTime returns the event time of the message.
 	EventTime() time.Time
+	// Watermark returns the watermark of the message.
 	Watermark() time.Time
+	// Headers returns the headers of the message.
 	Headers() map[string]string
 }