Rewrite gRPC service invocation example (dapr#883)

* original version of service invocation grpc proxying Signed-off-by: MregXN <[email protected]> * modify formatter and README Signed-off-by: MregXN <[email protected]> * inject grpc port automatically Signed-off-by: MregXN <[email protected]> * re-trigger validation Signed-off-by: MregXN <[email protected]> * use withInterceptors() as MetadataUtils.attachHeaders is deprecated Signed-off-by: MregXN <[email protected]> --------- Signed-off-by: MregXN <[email protected]> Signed-off-by: Artur Souza <[email protected]> Co-authored-by: Artur Souza <[email protected]> Co-authored-by: Mukundan Sundararajan <[email protected]> Co-authored-by: Cassie Coyle <[email protected]>
cicoyle · Dec 21, 2023 · 14d8363 · 14d8363
1 parent 49ccb31
commit 14d8363
Show file tree

Hide file tree

Showing 4 changed files with 186 additions and 178 deletions.
diff --git a/examples/proto/helloworld.proto b/examples/proto/helloworld.proto
@@ -7,13 +7,16 @@ option java_package = "io.dapr.examples";
 
 // User Code definitions
 service HelloWorld {
-  rpc Say (SayRequest) returns (SayResponse) {}
+  // Sends a greeting
+  rpc SayHello (HelloRequest) returns (HelloReply) {}
 }
 
-message SayRequest {
-  string message = 1;
+// The request message containing the user's name.
+message HelloRequest {
+  string name = 1;
 }
 
-message SayResponse {
-  string timestamp = 1;
+// The response message containing the greetings
+message HelloReply {
+  string message = 1;
 }
diff --git a/examples/src/main/java/io/dapr/examples/invoke/grpc/HelloWorldClient.java b/examples/src/main/java/io/dapr/examples/invoke/grpc/HelloWorldClient.java
@@ -13,43 +13,67 @@
 
 package io.dapr.examples.invoke.grpc;
 
-import io.dapr.client.DaprClient;
-import io.dapr.client.DaprClientBuilder;
-import io.dapr.client.domain.HttpExtension;
+import io.dapr.examples.DaprExamplesProtos.HelloReply;
+import io.dapr.examples.DaprExamplesProtos.HelloRequest;
+import io.dapr.examples.HelloWorldGrpc;
+import io.grpc.Grpc;
+import io.grpc.InsecureChannelCredentials;
+import io.grpc.ManagedChannel;
+import io.grpc.Metadata;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.MetadataUtils;
+
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  * 1. Build and install jars:
  * mvn clean install
  * 2. cd [repo root]/examples
  * 2. Send messages to the server:
- * dapr run -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.invoke.grpc.HelloWorldClient
+ * dapr run -- java -jar target/dapr-java-sdk-examples-exec.jar
+ * io.dapr.examples.invoke.grpc.HelloWorldClient
  */
 public class HelloWorldClient {
 
+  private static final Logger logger = Logger.getLogger(HelloWorldClient.class.getName());
+
   /**
    * The main method of the client app.
    *
    * @param args Array of messages to be sent.
    */
   public static void main(String[] args) throws Exception {
-    try (DaprClient client = new DaprClientBuilder().build()) {
 
-      String serviceAppId = "hellogrpc";
-      String method = "say";
+    String user = "World";
+    String target = "localhost:" + System.getenv("DAPR_GRPC_PORT");
+
+    ManagedChannel channel = Grpc.newChannelBuilder(target, InsecureChannelCredentials.create())
+        .build();
+
+    try {
+      HelloWorldGrpc.HelloWorldBlockingStub blockingStub = HelloWorldGrpc.newBlockingStub(channel);
 
-      int count = 0;
-      while (true) {
-        String message = "Message #" + (count++);
-        System.out.println("Sending message: " + message);
-        client.invokeMethod(serviceAppId, method, message, HttpExtension.NONE).block();
-        System.out.println("Message sent: " + message);
+      Metadata headers = new Metadata();
+      headers.put(Metadata.Key.of("dapr-app-id", Metadata.ASCII_STRING_MARSHALLER),
+          "hellogrpc");
 
-        Thread.sleep(1000);
+      // MetadataUtils.attachHeaders is deprecated.
+      blockingStub = blockingStub.withInterceptors(MetadataUtils.newAttachHeadersInterceptor(headers));
 
-        // This is an example, so for simplicity we are just exiting here.
-        // Normally a dapr app would be a web service and not exit main.
-        System.out.println("Done");
+      logger.info("Will try to greet " + user + " ...");
+      try {
+        HelloRequest request = HelloRequest.newBuilder().setName(user).build();
+        HelloReply response = blockingStub.sayHello(request);
+        logger.info("Greeting: " + response.getMessage());
+      } catch (StatusRuntimeException e) {
+        logger.log(Level.WARNING, "RPC failed: {0}", e.getStatus());
       }
+    } finally {
+      // To prevent leaking resources like threads and TCP connections
+      // the channel should be shut down when it will no longer be used.
+      channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
     }
   }
 }
diff --git a/examples/src/main/java/io/dapr/examples/invoke/grpc/HelloWorldService.java b/examples/src/main/java/io/dapr/examples/invoke/grpc/HelloWorldService.java
@@ -13,133 +13,112 @@
 
 package io.dapr.examples.invoke.grpc;
 
-import com.google.protobuf.Any;
-import io.dapr.v1.AppCallbackGrpc;
-import io.dapr.v1.CommonProtos;
+import io.dapr.examples.DaprExamplesProtos.HelloReply;
+import io.dapr.examples.DaprExamplesProtos.HelloRequest;
+import io.dapr.examples.HelloWorldGrpc;
+import io.grpc.Grpc;
+import io.grpc.InsecureServerCredentials;
 import io.grpc.Server;
-import io.grpc.ServerBuilder;
 import io.grpc.stub.StreamObserver;
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
 import org.apache.commons.cli.DefaultParser;
 import org.apache.commons.cli.Options;
 
 import java.io.IOException;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.Calendar;
-import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
 
-import static io.dapr.examples.DaprExamplesProtos.SayRequest;
-import static io.dapr.examples.DaprExamplesProtos.SayResponse;
 
 /**
  * 1. Build and install jars:
  * mvn clean install
  * 2. cd [repo root]/examples
  * 3. Run in server mode:
  * dapr run --app-id hellogrpc --app-port 5000 --app-protocol grpc \
- *   -- java -jar target/dapr-java-sdk-examples-exec.jar io.dapr.examples.invoke.grpc.HelloWorldService -p 5000
+ * -- java -jar target/dapr-java-sdk-examples-exec.jar
+ * io.dapr.examples.invoke.grpc.HelloWorldService -p 5000
  */
 public class HelloWorldService {
+  private static final Logger logger = Logger.getLogger(HelloWorldService.class.getName());
 
   /**
-   * Server mode: class that encapsulates all server-side logic for Grpc.
+   * Server mode: Grpc server.
    */
-  private static class GrpcHelloWorldDaprService extends AppCallbackGrpc.AppCallbackImplBase {
+  private Server server;
 
-    /**
-     * Format to output date and time.
-     */
-    private static final DateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
-
-    /**
-     * Server mode: Grpc server.
-     */
-    private Server server;
-
-    /**
-     * Server mode: starts listening on given port.
-     *
-     * @param port Port to listen on.
-     * @throws IOException Errors while trying to start service.
-     */
-    private void start(int port) throws IOException {
-      this.server = ServerBuilder
-          .forPort(port)
-          .addService(this)
-          .build()
-          .start();
-      System.out.printf("Server: started listening on port %d\n", port);
-
-      // Now we handle ctrl+c (or any other JVM shutdown)
-      Runtime.getRuntime().addShutdownHook(new Thread() {
-
-        @Override
-        public void run() {
-          System.out.println("Server: shutting down gracefully ...");
-          GrpcHelloWorldDaprService.this.server.shutdown();
-          System.out.println("Server: Bye.");
-        }
-      });
-    }
-
-    /**
-     * Server mode: waits for shutdown trigger.
-     *
-     * @throws InterruptedException Propagated interrupted exception.
-     */
-    private void awaitTermination() throws InterruptedException {
-      if (this.server != null) {
-        this.server.awaitTermination();
-      }
-    }
+  /**
+   * Server mode: class that encapsulates server-side handling logic for Grpc.
+   */
+  static class HelloWorldImpl extends HelloWorldGrpc.HelloWorldImplBase {
 
     /**
-     * Server mode: this is the Dapr method to receive Invoke operations via Grpc.
+     * Handling of the 'sayHello' method.
      *
-     * @param request          Dapr envelope request,
-     * @param responseObserver Dapr envelope response.
+     * @param request Request to say something.
+     * @return Response with when it was said.
      */
     @Override
-    public void onInvoke(CommonProtos.InvokeRequest request,
-                         StreamObserver<CommonProtos.InvokeResponse> responseObserver) {
-      try {
-        if ("say".equals(request.getMethod())) {
-          SayRequest sayRequest =
-              SayRequest.newBuilder().setMessage(request.getData().getValue().toStringUtf8()).build();
-          SayResponse sayResponse = this.say(sayRequest);
-          CommonProtos.InvokeResponse.Builder responseBuilder = CommonProtos.InvokeResponse.newBuilder();
-          responseBuilder.setData(Any.pack(sayResponse));
-          responseObserver.onNext(responseBuilder.build());
+    public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
+      logger.info("greet to " + req.getName());
+      HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
+      responseObserver.onNext(reply);
+      responseObserver.onCompleted();
+    }
+  }
+
+  /**
+   * Server mode: starts listening on given port.
+   *
+   * @param port Port to listen on.
+   * @throws IOException Errors while trying to start service.
+   */
+  private void start(int port) throws IOException {
+    server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
+        .addService(new HelloWorldImpl())
+        .build()
+        .start();
+    logger.info("Server started, listening on " + port);
+    Runtime.getRuntime().addShutdownHook(new Thread() {
+      @Override
+      public void run() {
+        // Use stderr here since the logger may have been reset by its JVM shutdown
+        // hook.
+        System.err.println("*** shutting down gRPC server since JVM is shutting down");
+        try {
+          HelloWorldService.this.stop();
+        } catch (InterruptedException e) {
+          e.printStackTrace(System.err);
         }
-      } finally {
-        responseObserver.onCompleted();
+        System.err.println("*** server shut down");
       }
-    }
-
-    /**
-     * Handling of the 'say' method.
-     *
-     * @param request Request to say something.
-     * @return Response with when it was said.
-     */
-    public SayResponse say(SayRequest request) {
-      Calendar utcNow = Calendar.getInstance(TimeZone.getTimeZone("GMT"));
-      String utcNowAsString = DATE_FORMAT.format(utcNow.getTime());
+    });
+  }
 
-      // Handles the request by printing message.
-      System.out.println("Server: " + request.getMessage());
-      System.out.println("@ " + utcNowAsString);
+  /**
+   * Server mode: waits for shutdown trigger.
+   *
+   * @throws InterruptedException Propagated interrupted exception.
+   */
+  private void stop() throws InterruptedException {
+    if (server != null) {
+      server.shutdown().awaitTermination(30, TimeUnit.SECONDS);
+    }
+  }
 
-      // Now respond with current timestamp.
-      SayResponse.Builder responseBuilder = SayResponse.newBuilder();
-      return responseBuilder.setTimestamp(utcNowAsString).build();
+  /**
+   * Await termination on the main thread since the grpc library uses daemon
+   * threads.
+   */
+  private void blockUntilShutdown() throws InterruptedException {
+    if (server != null) {
+      server.awaitTermination();
     }
   }
 
   /**
    * This is the main method of this app.
+   * 
    * @param args The port to listen on.
    * @throws Exception An Exception.
    */
@@ -153,8 +132,9 @@ public static void main(String[] args) throws Exception {
     // If port string is not valid, it will throw an exception.
     int port = Integer.parseInt(cmd.getOptionValue("port"));
 
-    final GrpcHelloWorldDaprService service = new GrpcHelloWorldDaprService();
+    final HelloWorldService service = new HelloWorldService();
     service.start(port);
-    service.awaitTermination();
+    service.blockUntilShutdown();
   }
+
 }