rename to transmission status

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

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

@@ -30,7 +30,7 @@ message MapRequest {
   // This ID is used to uniquely identify a map request
   string id = 2;
   optional Handshake handshake = 3;
-  optional Status status = 4;
+  optional TransmissionStatus status = 4;
 }
 
 /*
@@ -44,7 +44,7 @@ message Handshake {
 /*
  * Status message to indicate the status of the message.
  */
-message Status {
+message TransmissionStatus {
   bool eot = 1;
 }
 
@@ -61,7 +61,7 @@ message MapResponse {
   // This ID is used to refer the responses to the request it corresponds to.
   string id = 2;
   optional Handshake handshake = 3;
-  optional Status status = 4;
+  optional TransmissionStatus status = 4;
 }
 
 /**

pkg/batchmapper/service.go

@@ -170,7 +170,7 @@ func (fs *Service) processData(ctx context.Context, stream mappb.Map_MapFnServer
 	default:
 		// send the end of transmission message
 		eot := &mappb.MapResponse{
-			Status: &mappb.Status{
+			Status: &mappb.TransmissionStatus{
 				Eot: true,
 			},
 		}

pkg/batchmapper/service_test.go

@@ -133,7 +133,7 @@ func TestService_BatchMapFn(t *testing.T) {
 				},
 				{
 					Request: &mappb.MapRequest_Request{},
-					Status: &mappb.Status{
+					Status: &mappb.TransmissionStatus{
 						Eot: true,
 					},
 				},
@@ -163,7 +163,7 @@ func TestService_BatchMapFn(t *testing.T) {
 					Id: "test2",
 				},
 				{
-					Status: &mappb.Status{
+					Status: &mappb.TransmissionStatus{
 						Eot: true,
 					},
 				},
@@ -231,7 +231,7 @@ func TestService_BatchMapFn(t *testing.T) {
 					Id: "test2",
 				},
 				{
-					Status: &mappb.Status{
+					Status: &mappb.TransmissionStatus{
 						Eot: true,
 					},
 				},

pkg/mapstreamer/examples/flatmap_stream/main.go

@@ -13,6 +13,7 @@ type FlatMap struct {
 }
 
 func (f *FlatMap) MapStream(ctx context.Context, keys []string, d mapstreamer.Datum, messageCh chan<- mapstreamer.Message) {
+	// we have to close to indicate the end of the stream, otherwise the client will wait forever.
 	defer close(messageCh)
 	msg := d.Value()
 	_ = d.EventTime() // Event time is available

pkg/mapstreamer/service.go

@@ -102,7 +102,7 @@ func (fs *Service) writeResponseToClient(ctx context.Context, stream mappb.Map_M
 			if !ok {
 				// Send EOT message since we are done processing the request.
 				eotMessage := &mappb.MapResponse{
-					Status: &mappb.Status{Eot: true},
+					Status: &mappb.TransmissionStatus{Eot: true},
 					Id:     reqID,
 				}
 				if err := stream.Send(eotMessage); err != nil {

pkg/mapstreamer/service_test.go

@@ -82,6 +82,7 @@ func TestService_MapFn(t *testing.T) {
 		{
 			name: "map_fn_forward_msg",
 			handler: MapStreamerFunc(func(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message) {
+				defer close(messageCh)
 				msg := datum.Value()
 				messageCh <- NewMessage(msg).WithKeys([]string{keys[0] + "_test"})
 			}),
@@ -108,6 +109,7 @@ func TestService_MapFn(t *testing.T) {
 		{
 			name: "map_fn_forward_msg_forward_to_all",
 			handler: MapStreamerFunc(func(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message) {
+				defer close(messageCh)
 				msg := datum.Value()
 				messageCh <- NewMessage(msg)
 			}),
@@ -133,6 +135,7 @@ func TestService_MapFn(t *testing.T) {
 		{
 			name: "map_fn_forward_msg_drop_msg",
 			handler: MapStreamerFunc(func(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message) {
+				defer close(messageCh)
 				messageCh <- MessageToDrop()
 			}),
 			args: args{
@@ -204,6 +207,7 @@ func doHandshake(t *testing.T, stream proto.Map_MapFnClient) {
 func TestService_MapFn_SingleMessage_MultipleResponses(t *testing.T) {
 	svc := &Service{
 		MapperStream: MapStreamerFunc(func(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message) {
+			defer close(messageCh)
 			for i := 0; i < 10; i++ {
 				msg := fmt.Sprintf("response_%d", i)
 				messageCh <- NewMessage([]byte(msg)).WithKeys([]string{keys[0] + "_test"})
@@ -257,6 +261,7 @@ func TestService_MapFn_SingleMessage_MultipleResponses(t *testing.T) {
 func TestService_MapFn_Multiple_Messages(t *testing.T) {
 	svc := &Service{
 		MapperStream: MapStreamerFunc(func(ctx context.Context, keys []string, datum Datum, messageCh chan<- Message) {
+			defer close(messageCh)
 			msg := datum.Value()
 			messageCh <- NewMessage(msg).WithKeys([]string{keys[0] + "_test"})
 		}),