binder: Add unit tests for ClientInbound's message reassembly

Author: jdcormie

binder/src/main/java/io/grpc/binder/internal/BinderServer.java

@@ -70,6 +70,7 @@ public final class BinderServer implements InternalServer, LeakSafeOneWayBinder.
   private final LeakSafeOneWayBinder hostServiceBinder;
   private final BinderTransportSecurity.ServerPolicyChecker serverPolicyChecker;
   private final InboundParcelablePolicy inboundParcelablePolicy;
+  private final OneWayBinderProxy.Decorator clientBinderDecorator;
 
   @GuardedBy("this")
   private ServerListener listener;
@@ -92,6 +93,7 @@ private BinderServer(Builder builder) {
         ImmutableList.copyOf(checkNotNull(builder.streamTracerFactories, "streamTracerFactories"));
     this.serverPolicyChecker = BinderInternal.createPolicyChecker(builder.serverSecurityPolicy);
     this.inboundParcelablePolicy = builder.inboundParcelablePolicy;
+    this.clientBinderDecorator = builder.clientBinderDecorator;
     hostServiceBinder = new LeakSafeOneWayBinder(this);
   }
 
@@ -183,7 +185,7 @@ public synchronized boolean handleTransaction(int code, Parcel parcel) {
                   executorServicePool,
                   attrsBuilder.build(),
                   streamTracerFactories,
-                  OneWayBinderProxy.IDENTITY_DECORATOR,
+                  clientBinderDecorator,
                   callbackBinder);
           transport.start(listener.transportCreated(transport));
           return true;
@@ -225,6 +227,7 @@ public static class Builder {
         SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE);
     ServerSecurityPolicy serverSecurityPolicy = SecurityPolicies.serverInternalOnly();
     InboundParcelablePolicy inboundParcelablePolicy = InboundParcelablePolicy.DEFAULT;
+    OneWayBinderProxy.Decorator clientBinderDecorator = OneWayBinderProxy.IDENTITY_DECORATOR;
 
     public BinderServer build() {
       return new BinderServer(this);
@@ -295,5 +298,19 @@ public Builder setInboundParcelablePolicy(InboundParcelablePolicy inboundParcela
           checkNotNull(inboundParcelablePolicy, "inboundParcelablePolicy");
       return this;
     }
+
+    /**
+     * Sets the {@link OneWayBinderProxy.Decorator} to be applied to this server's "client Binders".
+     *
+     * <p>Tests can use this to capture post-setup transactions from server to client. The specified
+     * decorator will be applied every time a client connects. The decorated result will be used for
+     * all subsequent transactions to this client from the new ServerTransport.
+     *
+     * <p>Optional, {@link OneWayBinderProxy#IDENTITY_DECORATOR} is the default.
+     */
+    public Builder setClientBinderDecorator(OneWayBinderProxy.Decorator clientBinderDecorator) {
+      this.clientBinderDecorator = checkNotNull(clientBinderDecorator);
+      return this;
+    }
   }
 }

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

@@ -60,6 +60,7 @@
 import io.grpc.binder.internal.SettableAsyncSecurityPolicy.AuthRequest;
 import io.grpc.internal.AbstractTransportTest;
 import io.grpc.internal.ClientStream;
+import io.grpc.internal.ClientStreamListenerBase;
 import io.grpc.internal.ClientTransport;
 import io.grpc.internal.ClientTransportFactory.ClientTransportOptions;
 import io.grpc.internal.ConnectionClientTransport;
@@ -538,6 +539,98 @@ public void msgFragmentsDeliveredToServerOutOfOrder() throws Exception {
     assertThat(methodDescriptor.parseResponse(msg)).isEqualTo(largeMessage);
   }
 
+  @Test
+  public void singleTxnMsgsDeliveredToClientOutOfOrder() throws Exception {
+    server = newServerBuilder().setClientBinderDecorator(blockingDecorator).build();
+    registerServerWithRobolectric((BinderServer) server);
+    server.start(serverListener);
+
+    client = newClientTransport(server);
+    runIfNotNull(client.start(mockClientTransportListener));
+
+    QueueingOneWayBinderProxy queueingClientProxy =
+        new QueueingOneWayBinderProxy(takeNextBinder(blockingDecorator));
+    blockingDecorator.putNextResult(queueingClientProxy);
+
+    // Deliver the setup transaction without interference.
+    queueingClientProxy.deliver(takeNextTransaction(queueingClientProxy));
+    verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
+
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    ClientStream stream =
+        client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, noopTracers);
+    stream.start(clientStreamListener);
+    stream.request(2);
+
+    MockServerTransportListener serverTransportListener =
+        serverListener.takeListenerOrFail(TIMEOUT_MS, MILLISECONDS);
+    MockServerTransportListener.StreamCreation streamCreation =
+        serverTransportListener.takeStreamOrFail(TIMEOUT_MS, MILLISECONDS);
+
+    streamCreation.stream.writeMessage(methodDescriptor.streamResponse("one"));
+    streamCreation.stream.writeMessage(methodDescriptor.streamResponse("two"));
+    streamCreation.stream.flush();
+
+    // Expect one transaction from the server for each message.
+    QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingClientProxy);
+    QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingClientProxy);
+
+    // Deliver them to the client out of order!
+    queueingClientProxy.deliver(tx2);
+    queueingClientProxy.deliver(tx1);
+
+    // Client should deliver messages to the application in the order sent.
+    InputStream msg1 = takeNextMessage(clientStreamListener.messageQueue);
+    assertThat(methodDescriptor.parseResponse(msg1)).isEqualTo("one");
+    InputStream msg2 = takeNextMessage(clientStreamListener.messageQueue);
+    assertThat(methodDescriptor.parseResponse(msg2)).isEqualTo("two");
+  }
+
+  @Test
+  public void msgFragmentsDeliveredToClientOutOfOrder() throws Exception {
+    server = newServerBuilder().setClientBinderDecorator(blockingDecorator).build();
+    registerServerWithRobolectric((BinderServer) server);
+    server.start(serverListener);
+
+    client = newClientTransport(server);
+    runIfNotNull(client.start(mockClientTransportListener));
+
+    QueueingOneWayBinderProxy queueingClientProxy =
+        new QueueingOneWayBinderProxy(takeNextBinder(blockingDecorator));
+    blockingDecorator.putNextResult(queueingClientProxy);
+
+    // Deliver the setup transaction without interference.
+    queueingClientProxy.deliver(takeNextTransaction(queueingClientProxy));
+    verify(mockClientTransportListener, timeout(TIMEOUT_MS)).transportReady();
+
+    ClientStreamListenerBase clientStreamListener = new ClientStreamListenerBase();
+    ClientStream stream =
+        client.newStream(methodDescriptor, new Metadata(), CallOptions.DEFAULT, noopTracers);
+    stream.start(clientStreamListener);
+    stream.request(1);
+
+    MockServerTransportListener serverTransportListener =
+        serverListener.takeListenerOrFail(TIMEOUT_MS, MILLISECONDS);
+    MockServerTransportListener.StreamCreation streamCreation =
+        serverTransportListener.takeStreamOrFail(TIMEOUT_MS, MILLISECONDS);
+
+    String largeMessage = newStringOfLength(BlockPool.BLOCK_SIZE + 1);
+    streamCreation.stream.writeMessage(methodDescriptor.streamResponse(largeMessage));
+    streamCreation.stream.flush();
+
+    // Expect the client to split largeMessage into two transactions.
+    QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(queueingClientProxy);
+    QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(queueingClientProxy);
+
+    // Deliver them to the client out of order!
+    queueingClientProxy.deliver(tx2);
+    queueingClientProxy.deliver(tx1);
+
+    // Client should reassemble the message correctly.
+    InputStream msg = takeNextMessage(clientStreamListener.messageQueue);
+    assertThat(methodDescriptor.parseResponse(msg)).isEqualTo(largeMessage);
+  }
+
   private static OneWayBinderProxy takeNextBinder(
       BlockingBinderDecorator<OneWayBinderProxy> decorator) throws InterruptedException {
     OneWayBinderProxy proxy = decorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);