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

Signed-off-by: a3hadi <[email protected]>
numaproj · Jan 29, 2024 · d7ce392 · d7ce392
1 parent cb91f37
commit d7ce392
Show file tree

Hide file tree

Showing 4 changed files with 16 additions and 0 deletions.
diff --git a/src/main/java/io/numaproj/numaflow/sideinput/GRPCConfig.java b/src/main/java/io/numaproj/numaflow/sideinput/GRPCConfig.java
@@ -1,5 +1,6 @@
 package io.numaproj.numaflow.sideinput;
 
+import io.numaproj.numaflow.info.ServerInfoAccessor;
 import lombok.Builder;
 import lombok.Getter;
 
@@ -11,12 +12,14 @@
 public class GRPCConfig {
     private String socketPath;
     private int maxMessageSize;
+    private String infoFilePath;
 
     /**
      * Static method to create default GRPCConfig.
      */
     static GRPCConfig defaultGrpcConfig() {
         return GRPCConfig.newBuilder()
+                .infoFilePath(ServerInfoAccessor.DEFAULT_SERVER_INFO_FILE_PATH)
                 .maxMessageSize(Constants.DEFAULT_MESSAGE_SIZE)
                 .socketPath(Constants.DEFAULT_SOCKET_PATH).build();
     }

diff --git a/src/main/java/io/numaproj/numaflow/sideinput/Server.java b/src/main/java/io/numaproj/numaflow/sideinput/Server.java
@@ -1,8 +1,11 @@
 package io.numaproj.numaflow.sideinput;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.annotations.VisibleForTesting;
 import io.grpc.ServerBuilder;
 import io.numaproj.numaflow.shared.GrpcServerUtils;
+import io.numaproj.numaflow.info.ServerInfoAccessor;
+import io.numaproj.numaflow.info.ServerInfoAccessorImpl;
 import lombok.extern.slf4j.Slf4j;
 
 import java.util.concurrent.TimeUnit;
@@ -15,6 +18,7 @@ public class Server {
 
     private final GRPCConfig grpcConfig;
     private final Service service;
+    private final ServerInfoAccessor serverInfoAccessor = new ServerInfoAccessorImpl(new ObjectMapper());
     private io.grpc.Server server;
 
     /**
@@ -43,6 +47,10 @@ public Server(SideInputRetriever sideInputRetriever, GRPCConfig grpcConfig) {
      * @throws Exception if server fails to start
      */
     public void start() throws Exception {
+        GrpcServerUtils.writeServerInfo(
+                serverInfoAccessor,
+                grpcConfig.getSocketPath(),
+                grpcConfig.getInfoFilePath());
 
         if (this.server == null) {
             // create server builder

diff --git a/src/test/java/io/numaproj/numaflow/sideinput/GRPCConfigTest.java b/src/test/java/io/numaproj/numaflow/sideinput/GRPCConfigTest.java
@@ -1,5 +1,6 @@
 package io.numaproj.numaflow.sideinput;
 
+import io.numaproj.numaflow.info.ServerInfoAccessor;
 import org.junit.Assert;
 import org.junit.Test;
 
@@ -9,6 +10,9 @@ public class GRPCConfigTest {
     public void testDefaultGrpcConfig() {
         GRPCConfig grpcConfig = GRPCConfig.defaultGrpcConfig();
         Assert.assertNotNull(grpcConfig);
+        Assert.assertEquals(
+                ServerInfoAccessor.DEFAULT_SERVER_INFO_FILE_PATH,
+                grpcConfig.getInfoFilePath());
         Assert.assertEquals(Constants.DEFAULT_MESSAGE_SIZE, grpcConfig.getMaxMessageSize());
         Assert.assertEquals(Constants.DEFAULT_SOCKET_PATH, grpcConfig.getSocketPath());
     }

diff --git a/src/test/java/io/numaproj/numaflow/sideinput/ServerTest.java b/src/test/java/io/numaproj/numaflow/sideinput/ServerTest.java
@@ -28,6 +28,7 @@ public void setUp() throws Exception {
         GRPCConfig grpcServerConfig = GRPCConfig.newBuilder()
                 .maxMessageSize(Constants.DEFAULT_MESSAGE_SIZE)
                 .socketPath(Constants.DEFAULT_SOCKET_PATH)
+                .infoFilePath("/tmp/numaflow-test-server-info)")
                 .build();
 
         server = new Server(