binder: Add unit tests for ServerInbound message reassembly

Author: jdcormie

binder/src/test/java/io/grpc/binder/internal/RobolectricBinderTransportTest.java

@@ -47,15 +47,19 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.truth.TruthJUnit;
 import io.grpc.Attributes;
+import io.grpc.CallOptions;
 import io.grpc.InternalChannelz.SocketStats;
+import io.grpc.Metadata;
 import io.grpc.ServerStreamTracer;
 import io.grpc.Status;
 import io.grpc.binder.AndroidComponentAddress;
 import io.grpc.binder.ApiConstants;
 import io.grpc.binder.AsyncSecurityPolicy;
 import io.grpc.binder.SecurityPolicies;
+import io.grpc.binder.internal.OneWayBinderProxies.*;
 import io.grpc.binder.internal.SettableAsyncSecurityPolicy.AuthRequest;
 import io.grpc.internal.AbstractTransportTest;
+import io.grpc.internal.ClientStream;
 import io.grpc.internal.ClientTransport;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
 import io.grpc.internal.ConnectionClientTransport;
@@ -66,6 +70,7 @@
 import io.grpc.internal.MockServerTransportListener;
 import io.grpc.internal.ObjectPool;
 import io.grpc.internal.SharedResourcePool;
+import java.io.InputStream;
 import java.util.List;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
@@ -124,6 +129,8 @@ public final class RobolectricBinderTransportTest extends AbstractTransportTest
   ServiceInfo serviceInfo;
 
   private int nextServerAddress;
+  private BlockingBinderDecorator<OneWayBinderProxy> blockingDecorator =
+      new BlockingBinderDecorator<>();
 
   @Parameter(value = 0)
   public boolean preAuthServersParam;
@@ -433,4 +440,116 @@ public void flowControlPushBack() {}
   @Ignore("See BinderTransportTest#serverAlreadyListening")
   @Override
   public void serverAlreadyListening() {}
+
+  @Test
+  public void singleTxnMsgsDeliveredToServerOutOfOrder() throws Exception {
+    server.start(serverListener);
+    client = newClientTransportBuilder()
+        .setFactory(
+            newClientTransportFactoryBuilder()
+                .setBinderDecorator(blockingDecorator)
+                .buildClientTransportFactory())
+        .build();
+
+    runIfNotNull(client.start(mockClientTransportListener));
+
+    OneWayBinderProxy endpointProxy = blockingDecorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(endpointProxy).isNotNull();
+    blockingDecorator.putNextResult(endpointProxy);
+
+    OneWayBinderProxy serverProxy = blockingDecorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(serverProxy).isNotNull();
+    QueueingOneWayBinderProxy proxy = new QueueingOneWayBinderProxy(serverProxy);
+    blockingDecorator.putNextResult(proxy);
+
+    verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
+
+    ClientStream stream = client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, noopTracers);
+    stream.writeMessage(methodDescriptor.streamRequest("one"));
+    stream.writeMessage(methodDescriptor.streamRequest("two"));
+    stream.flush();
+
+    // We expect at least two transactions for the messages.
+    QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(proxy);
+    QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(proxy);
+
+    // Deliver them out of order!
+    proxy.deliver(tx2);
+    proxy.deliver(tx1);
+
+    // Verify that the server receives them in order.
+    MockServerTransportListener serverTransportListener = serverListener.takeListenerOrFail(TIMEOUT_MS, MILLISECONDS);
+    MockServerTransportListener.StreamCreation streamCreation = serverTransportListener.takeStreamOrFail(TIMEOUT_MS,
+        MILLISECONDS);
+    streamCreation.stream.request(2);
+
+    InputStream msg1 = streamCreation.listener.messageQueue.poll(TIMEOUT_MS, MILLISECONDS);
+    assertThat(msg1).isNotNull();
+    assertThat(methodDescriptor.parseResponse(msg1)).isEqualTo("one");
+
+    InputStream msg2 = streamCreation.listener.messageQueue.poll(TIMEOUT_MS, MILLISECONDS);
+    assertThat(msg2).isNotNull();
+    assertThat(methodDescriptor.parseResponse(msg2)).isEqualTo("two");
+  }
+
+  @Test
+  public void msgFragmentsDeliveredToServerOutOfOrder() throws Exception {
+    server.start(serverListener);
+    client = newClientTransportBuilder()
+        .setFactory(
+            newClientTransportFactoryBuilder()
+                .setBinderDecorator(blockingDecorator)
+                .buildClientTransportFactory())
+        .build();
+
+    runIfNotNull(client.start(mockClientTransportListener));
+
+    OneWayBinderProxy endpointProxy = blockingDecorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(endpointProxy).isNotNull();
+    blockingDecorator.putNextResult(endpointProxy);
+
+    OneWayBinderProxy serverProxy = blockingDecorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(serverProxy).isNotNull();
+    QueueingOneWayBinderProxy proxy = new QueueingOneWayBinderProxy(serverProxy);
+    blockingDecorator.putNextResult(proxy);
+
+    verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
+
+    ClientStream stream = client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, noopTracers);
+
+    String largeMessage = newStringOfLength(20 * 1024);
+    stream.writeMessage(methodDescriptor.streamRequest(largeMessage));
+    stream.flush();
+
+    // We expect at least two transactions for the large message.
+    QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(proxy);
+    QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(proxy);
+
+    // Deliver them out of order!
+    proxy.deliver(tx2);
+    proxy.deliver(tx1);
+
+    // Verify that the server receives the complete message correctly.
+    MockServerTransportListener serverTransportListener = serverListener.takeListenerOrFail(TIMEOUT_MS, MILLISECONDS);
+    MockServerTransportListener.StreamCreation streamCreation = serverTransportListener.takeStreamOrFail(TIMEOUT_MS,
+        MILLISECONDS);
+    streamCreation.stream.request(1);
+
+    InputStream msgStream = streamCreation.listener.messageQueue.poll(TIMEOUT_MS, MILLISECONDS);
+    assertThat(msgStream).isNotNull();
+    assertThat(methodDescriptor.parseResponse(msgStream)).isEqualTo(largeMessage);
+  }
+
+  private static QueueingOneWayBinderProxy.Transaction takeNextTransaction(
+      QueueingOneWayBinderProxy proxy) throws InterruptedException {
+    QueueingOneWayBinderProxy.Transaction tx = proxy.pollNextTransaction(TIMEOUT_MS, MILLISECONDS);
+    assertThat(tx).isNotNull();
+    return tx;
+  }
+
+  private static String newStringOfLength(int numChars) {
+    char[] chars = new char[numChars];
+    java.util.Arrays.fill(chars, 'x');
+    return new String(chars);
+  }
 }

core/src/testFixtures/java/io/grpc/internal/AbstractTransportTest.java

@@ -185,7 +185,7 @@ public void log(ChannelLogLevel level, String messageFormat, Object... args) {}
   protected final ClientStreamTracer[] tracers = new ClientStreamTracer[] {
       clientStreamTracer1, clientStreamTracer2
   };
-  private final ClientStreamTracer[] noopTracers = new ClientStreamTracer[] {
+  protected final ClientStreamTracer[] noopTracers = new ClientStreamTracer[] {
     new ClientStreamTracer() {}
   };
 
@@ -2195,7 +2195,7 @@ public void streamCreated(Attributes transportAttrs, Metadata metadata) {
     }
   }
 
-  private static class StringMarshaller implements MethodDescriptor.Marshaller<String> {
+  protected static class StringMarshaller implements MethodDescriptor.Marshaller<String> {
     public static final StringMarshaller INSTANCE = new StringMarshaller();
 
     @Override