also block onHalfClose after blocking onMessage

fixes #240

Author: jGleitz

grpc-spring-boot-starter/src/main/java/org/lognet/springboot/grpc/FailureHandlingServerInterceptor.java

@@ -1,10 +1,10 @@
 package org.lognet.springboot.grpc;
 
+import io.grpc.ForwardingServerCallListener;
 import io.grpc.Metadata;
 import io.grpc.ServerCall;
 import io.grpc.ServerInterceptor;
 import io.grpc.Status;
-import io.grpc.StatusRuntimeException;
 
 public interface FailureHandlingServerInterceptor extends ServerInterceptor {
     default  void closeCall(Object o, GRpcErrorHandler errorHandler, ServerCall<?, ?> call, Metadata headers, final Status status, Exception exception){
@@ -14,4 +14,25 @@ default  void closeCall(Object o, GRpcErrorHandler errorHandler, ServerCall<?, ?
         call.close(statusToSend, responseHeaders);
 
     }
+
+    class MessageBlockingServerCallListener<R> extends ForwardingServerCallListener.SimpleForwardingServerCallListener<R> {
+        private volatile boolean messageBlocked = false;
+
+        public MessageBlockingServerCallListener(ServerCall.Listener<R> delegate) {
+            super(delegate);
+        }
+
+        @Override
+        public void onHalfClose() {
+            // If the message was blocked, downstream never had a chance to react to it. Hence, the half-close signal would look like
+            // an error to them. So we do not propagate the signal in that case.
+            if (!messageBlocked) {
+                super.onHalfClose();
+            }
+        }
+
+        protected void blockMessage() {
+            messageBlocked = true;
+        }
+    }
 }

grpc-spring-boot-starter/src/main/java/org/lognet/springboot/grpc/security/SecurityInterceptor.java

@@ -1,5 +1,9 @@
 package org.lognet.springboot.grpc.security;
 
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Optional;
+
 import io.grpc.Context;
 import io.grpc.Contexts;
 import io.grpc.ForwardingServerCall;
@@ -22,10 +26,6 @@
 import org.springframework.security.core.context.SecurityContext;
 import org.springframework.security.core.context.SecurityContextHolder;
 
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.Optional;
-
 @Slf4j
 public class SecurityInterceptor extends AbstractSecurityInterceptor implements FailureHandlingServerInterceptor, Ordered {
 
@@ -193,9 +193,10 @@ private <RespT, ReqT> ServerCall.Listener<ReqT> fail(ServerCallHandler<ReqT, Res
 
 
         } else {
-            return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(next.startCall(call, headers)) {
+            return new MessageBlockingServerCallListener<ReqT>(next.startCall(call, headers)) {
                 @Override
                 public void onMessage(ReqT message) {
+                    blockMessage();
                     closeCall(message, errorHandler, call, headers, status, exception);
                 }
             };

grpc-spring-boot-starter/src/main/java/org/lognet/springboot/grpc/validation/ValidatingInterceptor.java

@@ -1,7 +1,12 @@
 package org.lognet.springboot.grpc.validation;
 
+import java.util.Optional;
+import java.util.Set;
+import javax.validation.ConstraintViolation;
+import javax.validation.ConstraintViolationException;
+import javax.validation.Validator;
+
 import io.grpc.ForwardingServerCall;
-import io.grpc.ForwardingServerCallListener;
 import io.grpc.Metadata;
 import io.grpc.ServerCall;
 import io.grpc.ServerCallHandler;
@@ -15,12 +20,6 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.core.Ordered;
 
-import javax.validation.ConstraintViolation;
-import javax.validation.ConstraintViolationException;
-import javax.validation.Validator;
-import java.util.Optional;
-import java.util.Set;
-
 
 public class ValidatingInterceptor implements FailureHandlingServerInterceptor, Ordered {
     private Validator validator;
@@ -53,14 +52,14 @@ public void sendMessage(RespT message) {
                 }
             }
         }, headers);
-        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(listener) {
+        return new MessageBlockingServerCallListener<ReqT>(listener) {
 
             @Override
             public void onMessage(ReqT message) {
                 final Set<ConstraintViolation<ReqT>> violations = validator.validate(message, RequestMessage.class);
                 if (!violations.isEmpty()) {
-                     closeCall(message,errorHandler,call,headers,Status.INVALID_ARGUMENT,new ConstraintViolationException(violations));
-
+                    blockMessage();
+                    closeCall(message,errorHandler,call,headers,Status.INVALID_ARGUMENT,new ConstraintViolationException(violations));
                 } else {
                     super.onMessage(message);
                 }