binder: Add unit tests for ClientInbound's message reassembly

Author: jdcormie

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;
@@ -547,6 +548,102 @@ public void msgFragmentsDeliveredToServerOutOfOrder() throws Exception {
     assertThat(methodDescriptor.parseResponse(msgStream)).isEqualTo(largeMessage);
   }
 
+  @Test
+  public void singleTxnMsgsDeliveredToClientOutOfOrder() throws Exception {
+    server = newServerBuilder().setClientBinderDecorator(blockingDecorator).build();
+    registerServerWithRobolectric((BinderServer) server);
+    server.start(serverListener);
+
+    client = newClientTransportBuilder().build();
+    runIfNotNull(client.start(mockClientTransportListener));
+
+    OneWayBinderProxy clientProxy = blockingDecorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(clientProxy).isNotNull();
+    QueueingOneWayBinderProxy proxy = new QueueingOneWayBinderProxy(clientProxy);
+    blockingDecorator.putNextResult(proxy);
+
+    // Deliver the handshake transaction!
+    QueueingOneWayBinderProxy.Transaction setupTx = takeNextTransaction(proxy);
+    proxy.deliver(setupTx);
+
+    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();
+
+    QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(proxy);
+    QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(proxy);
+
+    proxy.deliver(tx2);
+    proxy.deliver(tx1);
+
+    InputStream msg1 = clientStreamListener.messageQueue.poll(TIMEOUT_MS, MILLISECONDS);
+    assertThat(msg1).isNotNull();
+    assertThat(methodDescriptor.parseResponse(msg1)).isEqualTo("one");
+
+    InputStream msg2 = clientStreamListener.messageQueue.poll(TIMEOUT_MS, MILLISECONDS);
+    assertThat(msg2).isNotNull();
+    assertThat(methodDescriptor.parseResponse(msg2)).isEqualTo("two");
+  }
+
+  @Test
+  public void msgFragmentsDeliveredToClientOutOfOrder() throws Exception {
+    server = newServerBuilder().setClientBinderDecorator(blockingDecorator).build();
+    registerServerWithRobolectric((BinderServer) server);
+    server.start(serverListener);
+
+    client = newClientTransportBuilder().build();
+    runIfNotNull(client.start(mockClientTransportListener));
+
+    OneWayBinderProxy clientProxy = blockingDecorator.takeNextRequest(TIMEOUT_MS, MILLISECONDS);
+    assertThat(clientProxy).isNotNull();
+    QueueingOneWayBinderProxy proxy = new QueueingOneWayBinderProxy(clientProxy);
+    blockingDecorator.putNextResult(proxy);
+
+    // Deliver the handshake transaction!
+    QueueingOneWayBinderProxy.Transaction setupTx = takeNextTransaction(proxy);
+    proxy.deliver(setupTx);
+
+    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(20 * 1024);
+    streamCreation.stream.writeMessage(methodDescriptor.streamResponse(largeMessage));
+    streamCreation.stream.flush();
+
+    QueueingOneWayBinderProxy.Transaction tx1 = takeNextTransaction(proxy);
+    QueueingOneWayBinderProxy.Transaction tx2 = takeNextTransaction(proxy);
+
+    proxy.deliver(tx2);
+    proxy.deliver(tx1);
+
+    InputStream msgStream = clientStreamListener.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);