feat: Using gRPC Bidirectional Streaming for Map

pkg/apis/proto/map/v1/map.proto

@@ -9,7 +9,7 @@ package map.v1;
 
 service Map {
   // MapFn applies a function to each map request element.
-  rpc MapFn(MapRequest) returns (MapResponse);
+  rpc MapFn(stream MapRequest) returns (stream MapResponse);
 
   // IsReady is the heartbeat endpoint for gRPC.
   rpc IsReady(google.protobuf.Empty) returns (ReadyResponse);
@@ -19,11 +19,25 @@ service Map {
  * MapRequest represents a request element.
  */
 message MapRequest {
-  repeated string keys = 1;
-  bytes value = 2;
-  google.protobuf.Timestamp event_time = 3;
-  google.protobuf.Timestamp watermark = 4;
-  map<string, string> headers = 5;
+  message Request {
+    repeated string keys = 1;
+    bytes value = 2;
+    google.protobuf.Timestamp event_time = 3;
+    google.protobuf.Timestamp watermark = 4;
+    map<string, string> headers = 5;
+  }
+  Request request = 1;
+  // This ID is used to uniquely identify a map request
+  string id = 2;
+  optional Handshake handshake = 3;
+}
+
+/*
+ * Handshake message between client and server to indicate the start of transmission.
+ */
+message Handshake {
+  // Required field indicating the start of transmission.
+  bool sot = 1;
 }
 
 /**
@@ -36,6 +50,9 @@ message MapResponse {
     repeated string tags = 3;
   }
   repeated Result results = 1;
+  // This ID is used to refer the responses to the request it corresponds to.
+  string id = 2;
+  optional Handshake handshake = 3;
 }
 
 /**