refactor

Signed-off-by: Sidhant Kohli <[email protected]>

pkg/batchmapper/examples/batchmap-flatmap/main.go

@@ -14,19 +14,20 @@ func batchMapFn(_ context.Context, datums <-chan batchmapper.Datum) batchmapper.
 		msg := d.Value()
 		_ = d.EventTime() // Event time is available
 		_ = d.Watermark() // Watermark is available
-		results := batchmapper.NewBatchResponse(d.Id())
+		batchResponse := batchmapper.NewBatchResponse(d.Id())
 		strs := strings.Split(string(msg), ",")
 		for _, s := range strs {
-			results = results.Append(batchmapper.NewMessage([]byte(s)))
+			batchResponse = batchResponse.Append(batchmapper.NewMessage([]byte(s)))
 		}
-		batchResponses = batchResponses.Append(results)
+
+		batchResponses = batchResponses.Append(batchResponse)
 	}
 	return batchResponses
 }
 
 func main() {
 	err := batchmapper.NewServer(batchmapper.BatchMapperFunc(batchMapFn)).Start(context.Background())
 	if err != nil {
-		log.Panic("Failed to start map function server: ", err)
+		log.Panic("Failed to start batch map function server: ", err)
 	}
 }

pkg/batchmapper/message.go

@@ -6,7 +6,7 @@ var (
 	DROP = fmt.Sprintf("%U__DROP__", '\\') // U+005C__DROP__
 )
 
-// Message is used to wrap the data return by Map functions
+// Message is used to wrap the data return by Batch Map functions
 type Message struct {
 	value []byte
 	keys  []string
@@ -94,13 +94,13 @@ func BatchResponsesBuilder() BatchResponses {
 	return BatchResponses{}
 }
 
-// Append appends a Message
+// Append appends a batchResponse
 func (m BatchResponses) Append(msg batchResponse) BatchResponses {
 	m = append(m, msg)
 	return m
 }
 
-// Items returns the message list
+// Items returns the batchResponse list
 func (m BatchResponses) Items() []batchResponse {
 	return m
 }

pkg/batchmapper/server.go

@@ -50,7 +50,7 @@ func (m *server) Start(ctx context.Context) error {
 	defer log.Println("Successfully stopped the gRPC server")
 	defer grpcServer.GracefulStop()
 
-	// register the map service
+	// register the batch map service
 	batchmappb.RegisterBatchMapServer(grpcServer, m.svc)
 
 	// start the grpc server

pkg/batchmapper/service_test.go

@@ -95,7 +95,6 @@ func TestService_BatchMapFn(t *testing.T) {
 		input       []*batchmappb.BatchMapRequest
 		expected    []*batchmappb.BatchMapResponse
 		expectedErr bool
-		streamErr   bool
 	}{
 		{
 			name: "batch_map_stream_fn_forward_msg",
@@ -188,7 +187,6 @@ func TestService_BatchMapFn(t *testing.T) {
 				},
 			},
 			expectedErr: true,
-			streamErr:   true,
 		},
 	}
 	for _, tt := range tests {
@@ -206,7 +204,7 @@ func TestService_BatchMapFn(t *testing.T) {
 			result := make([]*batchmappb.BatchMapResponse, 0)
 
 			var udfBatchMapFnStream batchmappb.BatchMap_BatchMapFnServer
-			if tt.streamErr {
+			if tt.expectedErr {
 				udfBatchMapFnStream = NewBatchMapFnServerErrTest(ctx, inputCh, outputCh)
 			} else {
 				udfBatchMapFnStream = NewBatchBatchMapStreamFnServerTest(ctx, inputCh, outputCh)
@@ -236,8 +234,9 @@ func TestService_BatchMapFn(t *testing.T) {
 			close(inputCh)
 			wg.Wait()
 
-			if err != nil {
-				assert.True(t, tt.expectedErr, "BatchMapFn() error = %v, expectedErr %v", err, tt.expectedErr)
+			if tt.expectedErr {
+				// assert err is not nil
+				assert.NotNil(t, err)
 				return
 			}
 

pkg/sourcetransformer/server.go

@@ -49,7 +49,7 @@ func (m *server) Start(ctx context.Context) error {
 	defer log.Println("Successfully stopped the gRPC server")
 	defer grpcServer.GracefulStop()
 
-	// register the map service
+	// register the source transformer service
 	v1.RegisterSourceTransformServer(grpcServer, m.svc)
 
 	// start the grpc server