Merge remote-tracking branch 'origin/mcs-interop-tests' into mcs-interop-tests

Author: kannanjgithub

android-interop-testing/build.gradle

@@ -17,7 +17,6 @@ android {
                 srcDirs += "${projectDir}/../interop-testing/src/main/java/"
                 setIncludes(["io/grpc/android/integrationtest/**",
                              "io/grpc/testing/integration/AbstractInteropTest.java",
-                             "io/grpc/testing/integration/TestCases.java",
                              "io/grpc/testing/integration/TestServiceImpl.java",
                              "io/grpc/testing/integration/Util.java"])
             }

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceClient.java

@@ -1020,8 +1020,9 @@ public void testOrcaOob() throws Exception {
           .build();
 
       final int retryLimit = 5;
+      final Object lastItem = new Object();
       StreamingOutputCallResponseObserver streamingOutputCallResponseObserver =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver =
           asyncStub.fullDuplexCall(streamingOutputCallResponseObserver);
 
@@ -1054,7 +1055,7 @@ public void testOrcaOob() throws Exception {
       }
       assertThat(i).isLessThan(retryLimit);
       streamObserver.onCompleted();
-      assertThat(streamingOutputCallResponseObserver.verifiedCompleted()).isTrue();
+      assertThat(streamingOutputCallResponseObserver.take()).isSameInstanceAs(lastItem);
     }
 
     @Override
@@ -1084,9 +1085,13 @@ protected int operationTimeoutMillis() {
 
     class StreamingOutputCallResponseObserver implements
         StreamObserver<StreamingOutputCallResponse> {
-      private final Object lastItem = new Object();
+      private final Object lastItem;
       private final BlockingQueue<Object> queue = new LinkedBlockingQueue<>();
 
+      public StreamingOutputCallResponseObserver(Object lastItem) {
+        this.lastItem = lastItem;
+      }
+
       @Override
       public void onNext(StreamingOutputCallResponse value) {
         queue.add(value);
@@ -1105,59 +1110,56 @@ public void onCompleted() {
       Object take() throws InterruptedException {
         return queue.take();
       }
-
-      boolean verifiedCompleted() throws InterruptedException {
-        return queue.take() == lastItem;
-      }
     }
 
     public void testMcs(TestServiceGrpc.TestServiceStub asyncStub) throws Exception {
+      final Object lastItem = new Object();
       StreamingOutputCallResponseObserver responseObserver1 =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver1 =
           asyncStub.fullDuplexCall(responseObserver1);
       StreamingOutputCallRequest request = StreamingOutputCallRequest.newBuilder()
           .addResponseParameters(ResponseParameters.newBuilder()
-              .setSendClientSocketAddressInResponse(
+              .setFillPeerSocketAddress(
                   Messages.BoolValue.newBuilder().setValue(true).build())
               .build())
           .build();
       streamObserver1.onNext(request);
       Object responseObj = responseObserver1.take();
       StreamingOutputCallResponse callResponse = (StreamingOutputCallResponse) responseObj;
-      String clientSocketAddressInCall1 = callResponse.getClientSocketAddress();
+      String clientSocketAddressInCall1 = callResponse.getPeerSocketAddress();
       assertThat(clientSocketAddressInCall1).isNotEmpty();
 
       StreamingOutputCallResponseObserver responseObserver2 =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver2 =
           asyncStub.fullDuplexCall(responseObserver2);
       streamObserver2.onNext(request);
       callResponse = (StreamingOutputCallResponse) responseObserver2.take();
-      String clientSocketAddressInCall2 = callResponse.getClientSocketAddress();
+      String clientSocketAddressInCall2 = callResponse.getPeerSocketAddress();
 
       assertThat(clientSocketAddressInCall1).isEqualTo(clientSocketAddressInCall2);
 
       // The first connection is at max rpc call count of 2, so the 3rd rpc will cause a new
       // connection to be created in the same subchannel and not get queued.
       StreamingOutputCallResponseObserver responseObserver3 =
-          new StreamingOutputCallResponseObserver();
+          new StreamingOutputCallResponseObserver(lastItem);
       StreamObserver<StreamingOutputCallRequest> streamObserver3 =
           asyncStub.fullDuplexCall(responseObserver3);
       streamObserver3.onNext(request);
       callResponse = (StreamingOutputCallResponse) responseObserver3.take();
-      String clientSocketAddressInCall3 = callResponse.getClientSocketAddress();
+      String clientSocketAddressInCall3 = callResponse.getPeerSocketAddress();
 
       // This assertion is currently failing because connection scaling when MCS limit has been
       // reached is not yet implemented in gRPC Java.
       assertThat(clientSocketAddressInCall3).isNotEqualTo(clientSocketAddressInCall1);
 
       streamObserver1.onCompleted();
-      assertThat(responseObserver1.verifiedCompleted()).isTrue();
       streamObserver2.onCompleted();
-      assertThat(responseObserver2.verifiedCompleted()).isTrue();
       streamObserver3.onCompleted();
-      assertThat(responseObserver3.verifiedCompleted()).isTrue();
+      assertThat(responseObserver1.take()).isSameInstanceAs(lastItem);
+      assertThat(responseObserver2.take()).isSameInstanceAs(lastItem);
+      assertThat(responseObserver3.take()).isSameInstanceAs(lastItem);
     }
   }
 

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceImpl.java

@@ -245,7 +245,7 @@ public void onNext(StreamingOutputCallRequest request) {
         if (whetherSendClientSocketAddressInResponse(request)) {
           responseObserver.onNext(
               StreamingOutputCallResponse.newBuilder()
-                  .setClientSocketAddress(PEER_ADDRESS_CONTEXT_KEY.get().toString())
+                  .setPeerSocketAddress(PEER_ADDRESS_CONTEXT_KEY.get().toString())
                   .build());
           return;
         }
@@ -256,7 +256,7 @@ private boolean whetherSendClientSocketAddressInResponse(StreamingOutputCallRequ
         Iterator<ResponseParameters> responseParametersIterator =
             request.getResponseParametersList().iterator();
         while (responseParametersIterator.hasNext()) {
-          if (responseParametersIterator.next().getSendClientSocketAddressInResponse().getValue()) {
+          if (responseParametersIterator.next().getFillPeerSocketAddress().getValue()) {
             return true;
           }
         }

interop-testing/src/main/proto/grpc/testing/messages.proto

@@ -160,9 +160,9 @@ message ResponseParameters {
   // the response's compression status.
   BoolValue compressed = 3;
 
-  // Whether to request the server to send the requesting client's socket
+  // Whether to request the server to send the requesting peer's socket
   // address in the response.
-  BoolValue send_client_socket_address_in_response = 4;
+  BoolValue fill_peer_socket_address = 4;
 }
 
 // Server-streaming request.
@@ -191,8 +191,8 @@ message StreamingOutputCallResponse {
   // Payload to increase response size.
   Payload payload = 1;
 
-  // The client's socket address if requested.
-  string client_socket_address = 2;
+  // The peer's socket address if requested.
+  string peer_socket_address = 2;
 }
 
 // For reconnect interop test only.