xds: set end_of_stream on client half-close and handle end_of_stream_without_message per ext_proc spec

TAG=agy
CONV=86c74ebe-9fd7-4876-b27c-a4c1b230d346

Author: kannanjgithub

xds/src/main/java/io/grpc/xds/ExternalProcessorClientInterceptor.java

@@ -890,6 +890,7 @@ public void halfClose() {
       // Mode is GRPC
       sendToExtProc(ProcessingRequest.newBuilder()
           .setRequestBody(HttpBody.newBuilder()
+              .setEndOfStream(true)
               .setEndOfStreamWithoutMessage(true)
               .build())
           .build());
@@ -914,10 +915,13 @@ private void handleRequestBodyResponse(BodyResponse bodyResponse) {
         BodyMutation mutation = bodyResponse.getResponse().getBodyMutation();
         if (mutation.hasStreamedResponse()) {
           StreamedBodyResponse streamed = mutation.getStreamedResponse();
-          if (!streamed.getEndOfStreamWithoutMessage()) {
+          boolean isEndOfStream = streamed.getEndOfStream();
+          boolean isEndOfStreamWithoutMessage =
+              isEndOfStream && streamed.getEndOfStreamWithoutMessage();
+          if (!isEndOfStreamWithoutMessage) {
             super.sendMessage(new KnownLengthInputStream(streamed.getBody()));
           }
-          if (streamed.getEndOfStream() || streamed.getEndOfStreamWithoutMessage()) {
+          if (isEndOfStream) {
             if (requestSideClosed.compareAndSet(false, true)) {
               proceedWithHalfClose();
             }

xds/src/main/java/io/grpc/xds/ExternalProcessorServerInterceptor.java

@@ -1335,7 +1335,9 @@ int drainPendingMutatedRequestBodies() {
             break;
           }
           StreamedBodyResponse peeked = pendingMutatedRequestBodies.peek();
-          if (peeked.getEndOfStreamWithoutMessage()) {
+          boolean isEosWithoutMessage =
+              peeked.getEndOfStream() && peeked.getEndOfStreamWithoutMessage();
+          if (isEosWithoutMessage) {
             toDeliver.add(pendingMutatedRequestBodies.poll());
           } else if (pendingAppRequests.get() > 0) {
             pendingAppRequests.decrementAndGet();
@@ -1350,10 +1352,13 @@ int drainPendingMutatedRequestBodies() {
         final int bodySize = streamed.getBody().size();
         callContext.run(() -> {
           try {
-            if (!finalStreamed.getEndOfStreamWithoutMessage()) {
+            boolean isEndOfStream = finalStreamed.getEndOfStream();
+            boolean isEndOfStreamWithoutMessage =
+                isEndOfStream && finalStreamed.getEndOfStreamWithoutMessage();
+            if (!isEndOfStreamWithoutMessage) {
               wrappedListener.onExternalBody(finalStreamed.getBody());
             }
-            if (finalStreamed.getEndOfStream() || finalStreamed.getEndOfStreamWithoutMessage()) {
+            if (isEndOfStream) {
               wrappedListener.proceedWithHalfClose();
             }
           } finally {
@@ -1484,10 +1489,13 @@ private void drainRequestMessagesFailOpen() {
       for (StreamedBodyResponse streamed : mutatedToDeliver) {
         final StreamedBodyResponse finalStreamed = streamed;
         callContext.run(() -> {
-          if (!finalStreamed.getEndOfStreamWithoutMessage()) {
+          boolean isEndOfStream = finalStreamed.getEndOfStream();
+          boolean isEndOfStreamWithoutMessage =
+              isEndOfStream && finalStreamed.getEndOfStreamWithoutMessage();
+          if (!isEndOfStreamWithoutMessage) {
             wrappedListener.onExternalBody(finalStreamed.getBody());
           }
-          if (finalStreamed.getEndOfStream() || finalStreamed.getEndOfStreamWithoutMessage()) {
+          if (isEndOfStream) {
             wrappedListener.proceedWithHalfClose();
           }
         });
@@ -1908,6 +1916,7 @@ private void sendHalfCloseToExtProc() {
       }
 
       HttpBody.Builder bodyBuilder = HttpBody.newBuilder()
+          .setEndOfStream(true)
           .setEndOfStreamWithoutMessage(true);
 
       ProcessingRequest.Builder builder = ProcessingRequest.newBuilder()