chore: write server info for sideinput sdk (#97)

Signed-off-by: a3hadi <[email protected]>

pkg/shared/util.go

@@ -18,8 +18,6 @@ const (
 
 func PrepareServer(sockAddr string, infoFilePath string) (net.Listener, error) {
 	// If infoFilePath is not empty, write the server info to the file.
-	// For Side input we don't write data to the info server, hence will pass path as empty here.
-	// Could be used later on for similar cases
 	if infoFilePath != "" {
 		serverInfo := &info.ServerInfo{Protocol: info.UDS, Language: info.Go, Version: info.GetSDKVersion()}
 		if err := info.Write(serverInfo, info.WithServerInfoFilePath(infoFilePath)); err != nil {

pkg/sideinput/options.go

@@ -1,9 +1,12 @@
 package sideinput
 
+import "github.com/numaproj/numaflow-go/pkg/info"
+
 // options is the struct to hold the server options.
 type options struct {
-	sockAddr       string
-	maxMessageSize int
+	sockAddr           string
+	maxMessageSize     int
+	serverInfoFilePath string
 }
 
 // Option is the interface to apply options.
@@ -12,8 +15,9 @@ type Option func(*options)
 // defaultOptions returns the default options.
 func defaultOptions() *options {
 	return &options{
-		sockAddr:       address,
-		maxMessageSize: defaultMaxMessageSize,
+		sockAddr:           address,
+		maxMessageSize:     defaultMaxMessageSize,
+		serverInfoFilePath: info.ServerInfoFilePath,
 	}
 }
 
@@ -30,3 +34,10 @@ func WithSockAddr(addr string) Option {
 		opts.sockAddr = addr
 	}
 }
+
+// WithServerInfoFilePath sets the server info file path to the given path.
+func WithServerInfoFilePath(f string) Option {
+	return func(opts *options) {
+		opts.serverInfoFilePath = f
+	}
+}

pkg/sideinput/server.go

@@ -37,8 +37,7 @@ func (s *server) Start(ctx context.Context) error {
 	defer stop()
 
 	// start listening on unix domain socket
-	// For Side input we don't write data to the info server, hence will pass path as empty here.
-	lis, err := shared.PrepareServer(s.opts.sockAddr, "")
+	lis, err := shared.PrepareServer(s.opts.sockAddr, s.opts.serverInfoFilePath)
 	if err != nil {
 		return fmt.Errorf("failed to execute net.Listen(%q, %q): %v", uds, address, err)
 	}

pkg/sideinput/server_test.go

@@ -16,12 +16,17 @@ func TestServer_Start(t *testing.T) {
 		_ = os.RemoveAll(socketFile.Name())
 	}()
 
+	serverInfoFile, _ := os.CreateTemp("/tmp", "numaflow-test-info")
+	defer func() {
+		_ = os.RemoveAll(serverInfoFile.Name())
+	}()
+
 	var retrieveHandler = RetrieveFunc(func(ctx context.Context) Message {
 		return BroadcastMessage([]byte("test"))
 	})
 	// note: using actual uds connection
 	ctx, cancel := context.WithTimeout(context.Background(), 6*time.Second)
 	defer cancel()
-	err := NewSideInputServer(retrieveHandler, WithSockAddr(socketFile.Name())).Start(ctx)
+	err := NewSideInputServer(retrieveHandler, WithSockAddr(socketFile.Name()), WithServerInfoFilePath(serverInfoFile.Name())).Start(ctx)
 	assert.NoError(t, err)
 }