feature: udsink gRPC

function/start.go

@@ -1,5 +1,7 @@
-// Package golang provides an interface to write UDF in golang which will be exposed over HTTP. It accepts a handler of the following definition
-//  func(ctx context.Context, key, msg []byte) (messages Messages, err error)
+// Package function provides an interface to write UDF in golang which will be exposed over HTTP. It accepts a handler of the following definition
+//
+//	func(ctx context.Context, key, msg []byte) (messages Messages, err error)
+//
 // which will be invoked for message. If error is returned, the HTTP StatusCode will be set to 500.
 package function
 
@@ -139,7 +141,7 @@ func startWithContext(ctx context.Context, handler func(ctx context.Context, key
 	})
 
 	path := "/var/run/numaflow/udf.sock"
-	if err := os.Remove(path); !os.IsNotExist(err) && err != nil {
+	if err := os.RemoveAll(path); !os.IsNotExist(err) && err != nil {
 		return err
 	}
 	udsServer := &http.Server{}

go.mod

@@ -1,6 +1,6 @@
 module github.com/numaproj/numaflow-go
 
-go 1.18
+go 1.19
 
 require (
 	github.com/golang/mock v1.6.0

pkg/apis/proto/sink/v1/mockgen.go

@@ -0,0 +1,3 @@
+package v1
+
+//go:generate mockgen -destination sinkmock/sinkmock.go -package sinkmock github.com/numaproj/numaflow-go/pkg/apis/proto/sink/v1 UserDefinedSinkClient

pkg/apis/proto/sink/v1/udsink.proto

@@ -34,6 +34,7 @@ message Datum {
   bytes value = 2;
   EventTime event_time = 3;
   Watermark watermark = 4;
+  string id=5;
 }
 
 /**

pkg/function/examples/evenodd/main.go

@@ -4,12 +4,11 @@ import (
 	"context"
 	"strconv"
 
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow-go/pkg/function/server"
 )
 
-func handle(_ context.Context, key string, d datum.Datum) functionsdk.Messages {
+func handle(_ context.Context, key string, d functionsdk.Datum) functionsdk.Messages {
 	msg := d.Value()
 	_ = d.EventTime() // Event time is available
 	_ = d.Watermark() // Watermark is available

pkg/function/examples/flatmap/main.go

@@ -4,12 +4,11 @@ import (
 	"context"
 	"strings"
 
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow-go/pkg/function/server"
 )
 
-func handle(_ context.Context, key string, d datum.Datum) functionsdk.Messages {
+func handle(_ context.Context, key string, d functionsdk.Datum) functionsdk.Messages {
 	msg := d.Value()
 	_ = d.EventTime() // Event time is available
 	_ = d.Watermark() // Watermark is available

pkg/function/examples/forward_message/main.go

@@ -3,12 +3,11 @@ package main
 import (
 	"context"
 
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow-go/pkg/function/server"
 )
 
-func mapHandle(_ context.Context, key string, d datum.Datum) functionsdk.Messages {
+func mapHandle(_ context.Context, key string, d functionsdk.Datum) functionsdk.Messages {
 	// directly forward the input to the output
 	val := d.Value()
 	eventTime := d.EventTime()

pkg/function/examples/sum/main.go

@@ -5,12 +5,11 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow-go/pkg/function/server"
 )
 
-func reduceHandle(_ context.Context, key string, reduceCh <-chan datum.Datum, md functionsdk.Metadata) functionsdk.Messages {
+func reduceHandle(_ context.Context, key string, reduceCh <-chan functionsdk.Datum, md functionsdk.Metadata) functionsdk.Messages {
 	// sum up values for the same key
 	intervalWindow := md.IntervalWindow()
 	_ = intervalWindow

pkg/function/interface.go

@@ -5,16 +5,14 @@ import (
 	"time"
 
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
 
-// Client contains methods to call a gRPC client.
-type Client interface {
-	CloseConn(ctx context.Context) error
-	IsReady(ctx context.Context, in *emptypb.Empty) (bool, error)
-	MapFn(ctx context.Context, datum *functionpb.Datum) ([]*functionpb.Datum, error)
-	ReduceFn(ctx context.Context, datumStreamCh <-chan *functionpb.Datum) ([]*functionpb.Datum, error)
+// Datum contains methods to get the payload information.
+type Datum interface {
+	Value() []byte
+	EventTime() time.Time
+	Watermark() time.Time
 }
 
 // Metadata contains methods to get the metadata for the reduce operation.
@@ -28,29 +26,37 @@ type IntervalWindow interface {
 	EndTime() time.Time
 }
 
+// Client contains methods to call a gRPC client.
+type Client interface {
+	CloseConn(ctx context.Context) error
+	IsReady(ctx context.Context, in *emptypb.Empty) (bool, error)
+	MapFn(ctx context.Context, datum *functionpb.Datum) ([]*functionpb.Datum, error)
+	ReduceFn(ctx context.Context, datumStreamCh <-chan *functionpb.Datum) ([]*functionpb.Datum, error)
+}
+
 // MapHandler is the interface of map function implementation.
 type MapHandler interface {
 	// HandleDo is the function to process each coming message
-	HandleDo(ctx context.Context, key string, datum datum.Datum) Messages
+	HandleDo(ctx context.Context, key string, datum Datum) Messages
 }
 
 // ReduceHandler is the interface of reduce function implementation.
 type ReduceHandler interface {
-	HandleDo(ctx context.Context, key string, reduceCh <-chan datum.Datum, md Metadata) Messages
+	HandleDo(ctx context.Context, key string, reduceCh <-chan Datum, md Metadata) Messages
 }
 
 // MapFunc is utility type used to convert a HandleDo function to a MapHandler.
-type MapFunc func(ctx context.Context, key string, datum datum.Datum) Messages
+type MapFunc func(ctx context.Context, key string, datum Datum) Messages
 
 // HandleDo implements the function of map function.
-func (mf MapFunc) HandleDo(ctx context.Context, key string, datum datum.Datum) Messages {
+func (mf MapFunc) HandleDo(ctx context.Context, key string, datum Datum) Messages {
 	return mf(ctx, key, datum)
 }
 
 // ReduceFunc is utility type used to convert a HandleDo function to a ReduceHandler.
-type ReduceFunc func(ctx context.Context, key string, reduceCh <-chan datum.Datum, md Metadata) Messages
+type ReduceFunc func(ctx context.Context, key string, reduceCh <-chan Datum, md Metadata) Messages
 
 // HandleDo implements the function of reduce function.
-func (rf ReduceFunc) HandleDo(ctx context.Context, key string, reduceCh <-chan datum.Datum, md Metadata) Messages {
+func (rf ReduceFunc) HandleDo(ctx context.Context, key string, reduceCh <-chan Datum, md Metadata) Messages {
 	return rf(ctx, key, reduceCh, md)
 }

pkg/function/server/server_test.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	functionsdk "github.com/numaproj/numaflow-go/pkg/function"
 	"github.com/numaproj/numaflow-go/pkg/function/client"
 	"github.com/stretchr/testify/assert"
@@ -21,7 +20,7 @@ func Test_server_map(t *testing.T) {
 	file, err := os.CreateTemp("/tmp", "numaflow-test.sock")
 	assert.NoError(t, err)
 	defer func() {
-		err = os.Remove(file.Name())
+		err = os.RemoveAll(file.Name())
 		assert.NoError(t, err)
 	}()
 
@@ -36,7 +35,7 @@ func Test_server_map(t *testing.T) {
 		{
 			name: "server_map",
 			fields: fields{
-				mapHandler: functionsdk.MapFunc(func(ctx context.Context, key string, d datum.Datum) functionsdk.Messages {
+				mapHandler: functionsdk.MapFunc(func(ctx context.Context, key string, d functionsdk.Datum) functionsdk.Messages {
 					msg := d.Value()
 					return functionsdk.MessagesBuilder().Append(functionsdk.MessageTo(key+"_test", msg))
 				}),
@@ -83,7 +82,7 @@ func Test_server_reduce(t *testing.T) {
 	file, err := os.CreateTemp("/tmp", "numaflow-test.sock")
 	assert.NoError(t, err)
 	defer func() {
-		err = os.Remove(file.Name())
+		err = os.RemoveAll(file.Name())
 		assert.NoError(t, err)
 	}()
 
@@ -100,7 +99,7 @@ func Test_server_reduce(t *testing.T) {
 		{
 			name: "server_reduce",
 			fields: fields{
-				reduceHandler: functionsdk.ReduceFunc(func(ctx context.Context, key string, reduceCh <-chan datum.Datum, md functionsdk.Metadata) functionsdk.Messages {
+				reduceHandler: functionsdk.ReduceFunc(func(ctx context.Context, key string, reduceCh <-chan functionsdk.Datum, md functionsdk.Metadata) functionsdk.Messages {
 					// sum up values for the same key
 
 					// in this test case, md is nil

pkg/function/service.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/emptypb"
 )
@@ -84,7 +83,7 @@ func (fs *Service) ReduceFn(stream functionpb.UserDefinedFunction_ReduceFnServer
 	var (
 		ctx      = stream.Context()
 		key      string
-		reduceCh = make(chan datum.Datum)
+		reduceCh = make(chan Datum)
 		md       Metadata
 	)
 

pkg/function/service_test.go

@@ -7,7 +7,6 @@ import (
 	"time"
 
 	functionpb "github.com/numaproj/numaflow-go/pkg/apis/proto/function/v1"
-	"github.com/numaproj/numaflow-go/pkg/datum"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
@@ -32,7 +31,7 @@ func TestService_MapFn(t *testing.T) {
 		{
 			name: "map_fn_forward_msg",
 			fields: fields{
-				Mapper: MapFunc(func(ctx context.Context, key string, datum datum.Datum) Messages {
+				Mapper: MapFunc(func(ctx context.Context, key string, datum Datum) Messages {
 					msg := datum.Value()
 					return MessagesBuilder().Append(MessageTo(key+"_test", msg))
 				}),