Bael 7709

grpc/pom.xml

@@ -79,7 +79,7 @@
     </build>
 
     <properties>
-        <io.grpc.version>1.62.2</io.grpc.version>
+        <io.grpc.version>1.63.0</io.grpc.version>
         <protoc.version>3.17.2</protoc.version>
         <os-maven-plugin.version>1.6.2</os-maven-plugin.version>
         <protobuf-maven-plugin.version>0.6.1</protobuf-maven-plugin.version>

grpc/src/main/java/com/baeldung/grpc/orderprocessing/orderprocessing/GlobalExceptionInterceptor.java

@@ -0,0 +1,60 @@
+package com.baeldung.grpc.orderprocessing.orderprocessing;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.grpc.ForwardingServerCallListener;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+import io.grpc.Status;
+
+public class GlobalExceptionInterceptor implements ServerInterceptor {
+    private static final Logger logger = LoggerFactory.getLogger(GlobalExceptionInterceptor.class);
+
+    @Override
+    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall, Metadata headers,
+        ServerCallHandler<ReqT, RespT> next) {
+        ServerCall.Listener<ReqT> delegate = null;
+        logger.info("In the interceptor, going to call the target RPC");
+        try {
+            delegate = next.startCall(serverCall, headers);
+        } catch(Exception ex) {
+            logger.error("Exception caught in a subsequent interceptor", ex.getMessage());
+            return handleInterceptorException(ex, serverCall);
+        }
+        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(delegate) {
+            @Override
+            public void onHalfClose() {
+                try {
+                    super.onHalfClose();
+                } catch (Exception ex) {
+                    logger.error("Exception caught in a RPC endpoint", ex.getMessage());
+                    handleEndpointException(ex, serverCall);
+
+                }
+            }
+        };
+    }
+
+    private static <ReqT, RespT> void handleEndpointException(Exception ex, ServerCall<ReqT, RespT> serverCall) {
+        logger.error("The ex.getMessage() = {}", ex.getMessage());
+        String ticket = new TicketService().createTicket(ex.getMessage());
+        //send the response back
+        serverCall.close(Status.INTERNAL
+            .withCause(ex)
+            .withDescription(ex.getMessage() + ", Ticket raised:" + ticket), new Metadata());
+    }
+
+    private <ReqT, RespT> ServerCall.Listener<ReqT> handleInterceptorException(Throwable t, ServerCall<ReqT, RespT> serverCall) {
+        String ticket = new TicketService().createTicket(t.getMessage());
+        serverCall.close(Status.INTERNAL
+            .withCause(t)
+            .withDescription("An exception occurred in a **subsequent** interceptor:" + ", Ticket raised:" + ticket), new Metadata());
+
+        return new ServerCall.Listener<ReqT>() {
+            // no-op
+        };
+    }
+}

grpc/src/main/java/com/baeldung/grpc/orderprocessing/orderprocessing/LogInterceptor.java

@@ -0,0 +1,27 @@
+package com.baeldung.grpc.orderprocessing.orderprocessing;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+
+public class LogInterceptor implements ServerInterceptor {
+    private static final Logger logger = LoggerFactory.getLogger(LogInterceptor.class);
+
+    @Override
+    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(ServerCall<ReqT, RespT> serverCall, Metadata metadata,
+        ServerCallHandler<ReqT, RespT> next) {
+        logger.info("Entering LogInterceptor");
+        logMessage(serverCall);
+        ServerCall.Listener<ReqT> delegate = next.startCall(serverCall, metadata);
+        return delegate;
+    }
+
+    private <ReqT, RespT> void logMessage(ServerCall<ReqT, RespT> call) {
+        //forcing exception
+        int result = 100/0;
+    }
+}

grpc/src/main/java/com/baeldung/grpc/orderprocessing/orderprocessing/OrderProcessorImpl.java

@@ -0,0 +1,42 @@
+package com.baeldung.grpc.orderprocessing.orderprocessing;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.baeldung.grpc.orderprocessing.OrderProcessorGrpc;
+import com.baeldung.grpc.orderprocessing.OrderRequest;
+import com.baeldung.grpc.orderprocessing.OrderResponse;
+
+import io.grpc.Status;
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+
+public class OrderProcessorImpl extends OrderProcessorGrpc.OrderProcessorImplBase {
+    private static final Logger logger = LoggerFactory.getLogger(OrderProcessorImpl.class);
+
+    @Override
+    public void createOrder(OrderRequest request, StreamObserver<OrderResponse> responseObserver) {
+        if (!validateOrder(request)) {
+             throw new StatusRuntimeException(Status.FAILED_PRECONDITION.withDescription("Order Validation failed"));
+        } else {
+            OrderResponse orderResponse = processOrder(request);
+
+            responseObserver.onNext(orderResponse);
+            responseObserver.onCompleted();
+        }
+    }
+
+    private boolean validateOrder(OrderRequest request) {
+        //forcing an exception
+        int tax = 100/0;
+        return false;
+    }
+
+    private OrderResponse processOrder(OrderRequest request) {
+        logger.info("place the order with quantity:" + request.getQuantity());
+        return OrderResponse.newBuilder()
+            .setOrderID("ORD-5566")
+            .setResponse("Order placed successfully")
+            .build();
+    }
+}

grpc/src/main/java/com/baeldung/grpc/orderprocessing/orderprocessing/TicketService.java

@@ -0,0 +1,16 @@
+package com.baeldung.grpc.orderprocessing.orderprocessing;
+
+import java.util.Random;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TicketService {
+    private static final Logger logger = LoggerFactory.getLogger(TicketService.class);
+
+    public String createTicket(String error) {
+        logger.info("Lets create ticket");
+        return "TKT-" + new Random().nextInt();
+    }
+
+}

grpc/src/main/java/com/baeldung/grpc/orderprocessing/server/OrderProcessingServer.java

@@ -0,0 +1,22 @@
+package com.baeldung.grpc.orderprocessing.server;
+
+import java.io.IOException;
+
+import com.baeldung.grpc.orderprocessing.orderprocessing.GlobalExceptionInterceptor;
+import com.baeldung.grpc.orderprocessing.orderprocessing.LogInterceptor;
+import com.baeldung.grpc.orderprocessing.orderprocessing.OrderProcessorImpl;
+
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+
+public class OrderProcessingServer {
+    public static void main(String[] args) throws IOException, InterruptedException {
+        Server server = ServerBuilder.forPort(8080)
+            .addService(new OrderProcessorImpl())
+            .intercept(new LogInterceptor())
+            .intercept(new GlobalExceptionInterceptor())
+            .build();
+        server.start();
+        server.awaitTermination();
+    }
+}

grpc/src/main/proto/order_processing.proto

@@ -0,0 +1,22 @@
+syntax = "proto3";
+
+package orderprocessing;
+
+option java_multiple_files = true;
+option java_package = "com.baeldung.grpc.orderprocessing";
+
+message OrderRequest {
+  string product = 1;
+  int32 quantity = 2;
+  float price = 3;
+  string orderID = 4;
+}
+message OrderResponse {
+  string response = 1;
+  string orderID = 2;
+  string error = 3;
+}
+service OrderProcessor {
+  rpc createOrder(OrderRequest) returns (OrderResponse){}
+  rpc getOrder(OrderRequest) returns (OrderResponse){}
+}

grpc/src/main/resources/order_processing_cld.puml

@@ -0,0 +1,62 @@
+@startuml
+'https://plantuml.com/sequence-diagram
+
+hide empty attributes
+skinparam Handwritten false
+skinparam ClassBorderColor black
+skinparam BackgroundColor #fffce8/#f8f9fa
+skinparam class {
+ ArrowColor SeaGreen
+ BackgroundColor #fffce8
+}
+class OrderProcessorGrpc {
+}
+class OrderProcessorImplBase {
+   +createOrder()
+}
+class ServerBuilder {
++intercept()
++addService()
+}
+class OrderProcessorImpl {
+   +createOrder()
+   -validateOrder()
+}
+class OrderRequest {
+   +setOrderID()
+   +setProductID()
+   +setQuantity();
+}
+class OrderResponse {
+   +setOrderID()
+   +setResponse()
+   +setError()
+}
+class GlobalExceptionInterceptor {
+   +interceptCall()
+   -handleEndpointException()
+   -handleInterceptorException()
+}
+class LogInterceptor {
+   +interceptCall()
+   -logMessage()
+}
+class TicketService {
+   +createTicket()
+}
+
+class ServerInterceptor {
++interceptCall()
+}
+GlobalExceptionInterceptor --> TicketService: uses
+GlobalExceptionInterceptor --|> ServerInterceptor: implements
+LogInterceptor --|> ServerInterceptor: implements
+OrderProcessorImpl -left-> OrderRequest: uses
+OrderProcessorImpl -down-> OrderResponse: uses
+OrderProcessorImpl -up-> OrderProcessorImplBase: extends
+OrderProcessorImplBase-left-+ OrderProcessorGrpc
+ServerBuilder --> OrderProcessorImpl: uses
+ServerBuilder --> GlobalExceptionInterceptor: uses
+ServerBuilder --> LogInterceptor: uses
+
+@enduml

grpc/src/test/java/com/baeldung/grpc/orderprocessing/LogInterceptorUnitTest.java

@@ -0,0 +1,64 @@
+package com.baeldung.grpc.orderprocessing;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.Rule;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.baeldung.grpc.orderprocessing.orderprocessing.GlobalExceptionInterceptor;
+import com.baeldung.grpc.orderprocessing.orderprocessing.LogInterceptor;
+import com.baeldung.grpc.orderprocessing.orderprocessing.OrderProcessorImpl;
+
+import io.grpc.StatusRuntimeException;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.testing.GrpcCleanupRule;
+
+public class LogInterceptorUnitTest {
+
+    private static final Logger logger = LoggerFactory.getLogger(OrderProcessingUnitTest.class);
+    @Rule
+    public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
+    OrderProcessorGrpc.OrderProcessorBlockingStub orderProcessorBlockingStub;
+
+    @BeforeEach
+    public void setup() throws Exception {
+
+        String serverName = InProcessServerBuilder.generateName();
+
+        grpcCleanup.register(InProcessServerBuilder.forName(serverName)
+            .directExecutor()
+            .addService(new OrderProcessorImpl())
+            .intercept(new LogInterceptor())
+            .intercept(new GlobalExceptionInterceptor())
+            .build()
+            .start());
+
+        orderProcessorBlockingStub = OrderProcessorGrpc.newBlockingStub(grpcCleanup.register(
+            InProcessChannelBuilder.forName(serverName)
+                .directExecutor()
+                .build()));
+    }
+
+    @Test
+    void whenRuntimeExceptionInLogInterceptor_thenHandleException() {
+        OrderRequest orderRequest = OrderRequest.newBuilder()
+            .setProduct("PRD-7788")
+            .setQuantity(1)
+            .setPrice(5000)
+            .build();
+
+        try {
+            OrderResponse response = orderProcessorBlockingStub.createOrder(orderRequest);
+        } catch (StatusRuntimeException ex) {
+            assertTrue(ex.getStatus()
+                .getDescription()
+                .contains("An exception occurred in a **subsequent** interceptor:, Ticket raised:TKT"));
+            logger.error("Error: {}", ex.getStatus().getDescription());
+        }
+        logger.info("order processing over");
+    }
+}

grpc/src/test/java/com/baeldung/grpc/orderprocessing/OrderProcessingUnitTest.java

@@ -0,0 +1,63 @@
+package com.baeldung.grpc.orderprocessing;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.junit.Rule;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.baeldung.grpc.orderprocessing.orderprocessing.GlobalExceptionInterceptor;
+import com.baeldung.grpc.orderprocessing.orderprocessing.OrderProcessorImpl;
+
+import io.grpc.StatusRuntimeException;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.testing.GrpcCleanupRule;
+
+public class OrderProcessingUnitTest {
+
+    private static final Logger logger = LoggerFactory.getLogger(OrderProcessingUnitTest.class);
+    @Rule
+    public final GrpcCleanupRule grpcCleanup = new GrpcCleanupRule();
+    OrderProcessorGrpc.OrderProcessorBlockingStub orderProcessorBlockingStub;
+
+    @BeforeEach
+    public void setup() throws Exception {
+
+        String serverName = InProcessServerBuilder.generateName();
+
+        grpcCleanup.register(InProcessServerBuilder.forName(serverName)
+            .directExecutor()
+            .addService(new OrderProcessorImpl())
+            .intercept(new GlobalExceptionInterceptor())
+            .build()
+            .start());
+
+        orderProcessorBlockingStub = OrderProcessorGrpc.newBlockingStub(grpcCleanup.register(
+            InProcessChannelBuilder.forName(serverName)
+                .directExecutor()
+                .build()));
+    }
+
+    @Test
+    void whenRuntimeExceptionInRPCEndpoint_thenHandleException() {
+        OrderRequest orderRequest = OrderRequest.newBuilder()
+            .setProduct("PRD-7788")
+            .setQuantity(1)
+            .setPrice(5000)
+            .build();
+
+        try {
+            OrderResponse response = orderProcessorBlockingStub.createOrder(orderRequest);
+        } catch (StatusRuntimeException ex) {
+            assertTrue(ex.getStatus()
+                .getDescription()
+                .contains("Ticket raised:TKT"));
+            logger.error("Error: {}", ex.getStatus().getDescription());
+        }
+        logger.info("order processing over");
+    }
+}
+