When the target of your bindService() call is hosted in a different process, Android supplies * you with an {@link IBinder} that proxies your transactions to the remote {@link @@ -22,8 +24,8 @@ * consequences with respect to reentrancy, locking, and transaction dispatch order can be * surprising and dangerous. * - *
Wrap your {@link IBinder}s with an instance of this class to ensure the following - * out-of-process "oneway" semantics are always in effect: + *
Implementations of this interface ensure the following out-of-process "oneway" semantics are + * always in effect: * *
NB: One difference that this class can't conceal is that calls to onTransact() are serialized - * per {@link OneWayBinderProxy} instance, not per instance of the wrapped {@link IBinder}. An - * android.os.Binder with in-process callers could still receive concurrent calls to onTransact() on - * different threads if callers used different {@link OneWayBinderProxy} instances or if that Binder - * also had out-of-process callers. */ public abstract class OneWayBinderProxy { - private static final Logger logger = Logger.getLogger(OneWayBinderProxy.class.getName()); - protected final IBinder delegate; - - protected OneWayBinderProxy(IBinder iBinder) { - this.delegate = iBinder; - } /** * Returns a new instance of {@link OneWayBinderProxy} that wraps {@code iBinder}. * + *
NB: One difference this wrapper can't conceal is that calls to onTransact() are serialized + * per {@link OneWayBinderProxy} instance, not per instance of the wrapped {@link IBinder}. An + * android.os.Binder with in-process callers could still receive concurrent calls to onTransact() + * on different threads if callers used different {@link OneWayBinderProxy} instances or if that + * Binder also had out-of-process callers. + * * @param iBinder the binder to wrap * @param inProcessThreadHopExecutor a non-direct Executor used to dispatch calls to onTransact(), * if necessary @@ -81,7 +77,7 @@ public interface Decorator { public static final Decorator IDENTITY_DECORATOR = (x) -> x; /** - * Enqueues a transaction for the wrapped {@link IBinder} with guaranteed "oneway" semantics. + * Enqueues a transaction for the recipient with guaranteed "oneway" semantics. * *
NB: Unlike {@link IBinder#transact}, implementations of this method take ownership of the
* {@code data} Parcel. When this method returns, {@code data} will normally be empty, but callers
@@ -96,14 +92,47 @@ public interface Decorator {
public abstract void transact(int code, ParcelHolder data) throws RemoteException;
/**
- * Returns the wrapped {@link IBinder} for the purpose of calling methods other than {@link
- * IBinder#transact(int, Parcel, Parcel, int)}.
+ * Registers a death recipient to be notified when the host process of the remote binder dies.
+ *
+ * @see IBinder#linkToDeath(IBinder.DeathRecipient, int)
*/
- public IBinder getDelegate() {
- return delegate;
+ public abstract void linkToDeath(IBinder.DeathRecipient recipient, int flags)
+ throws RemoteException;
+
+ /**
+ * Unregisters a previously registered death recipient.
+ *
+ * @see IBinder#unlinkToDeath(IBinder.DeathRecipient, int)
+ */
+ public abstract boolean unlinkToDeath(IBinder.DeathRecipient recipient, int flags);
+
+ abstract static class WrappingImplBase extends OneWayBinderProxy {
+ protected final IBinder delegate;
+
+ WrappingImplBase(IBinder delegate) {
+ this.delegate = checkNotNull(delegate);
+ }
+
+ @Override
+ public void linkToDeath(IBinder.DeathRecipient recipient, int flags) throws RemoteException {
+ delegate.linkToDeath(recipient, flags);
+ }
+
+ @Override
+ public boolean unlinkToDeath(IBinder.DeathRecipient recipient, int flags) {
+ return delegate.unlinkToDeath(recipient, flags);
+ }
+
+ protected boolean transactAndRecycleParcel(int code, Parcel data) throws RemoteException {
+ try {
+ return delegate.transact(code, data, null, IBinder.FLAG_ONEWAY);
+ } finally {
+ data.recycle();
+ }
+ }
}
- static class OutOfProcessImpl extends OneWayBinderProxy {
+ static class OutOfProcessImpl extends WrappingImplBase {
OutOfProcessImpl(IBinder iBinder) {
super(iBinder);
}
@@ -118,15 +147,8 @@ public void transact(int code, ParcelHolder data) throws RemoteException {
}
}
- protected boolean transactAndRecycleParcel(int code, Parcel data) throws RemoteException {
- try {
- return delegate.transact(code, data, null, IBinder.FLAG_ONEWAY);
- } finally {
- data.recycle();
- }
- }
-
- static class InProcessImpl extends OneWayBinderProxy {
+ static class InProcessImpl extends WrappingImplBase {
+ private static final Logger logger = Logger.getLogger(InProcessImpl.class.getName());
private final SerializingExecutor executor;
InProcessImpl(IBinder binder, Executor executor) {
diff --git a/binder/src/test/java/io/grpc/binder/internal/BinderServerTransportTest.java b/binder/src/test/java/io/grpc/binder/internal/BinderServerTransportTest.java
index d261ce43c8c..a8fbec75a5b 100644
--- a/binder/src/test/java/io/grpc/binder/internal/BinderServerTransportTest.java
+++ b/binder/src/test/java/io/grpc/binder/internal/BinderServerTransportTest.java
@@ -29,12 +29,9 @@
import android.os.RemoteException;
import com.google.common.collect.ImmutableList;
import io.grpc.Attributes;
-import io.grpc.ServerStreamTracer;
import io.grpc.Status;
import io.grpc.internal.FixedObjectPool;
import io.grpc.internal.MockServerTransportListener;
-import io.grpc.internal.ObjectPool;
-import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import org.junit.Before;
import org.junit.Rule;
@@ -125,45 +122,4 @@ public void testStartAfterShutdownNoIdle() throws Exception {
assertThat(transportListener.isTerminated()).isTrue();
}
-
- static class BinderServerTransportBuilder {
- ObjectPool Both ClientInbound and ServerInbound share certain functionality from Inbound like message
+ * reassembly and flow control. This file tests both {@link Inbound.ClientInbound} specifics and the
+ * functionality common to both Inbounds to avoid duplicating tests in {@link ServerInboundTest}.
+ *
+ * Threading model: All Executors lead to Robolectric's main thread, where everything runs,
+ * including the test cases themselves. This makes it easy to write deterministic positive and
+ * negative assertions about listener callbacks because we can drain all executors and know that if
+ * the SUT was going to do something, it would have already happened. It certainly isn't realistic
+ * with respect to concurrency but that aspect is integration tested elsewhere (at a higher level).
+ */
+@RunWith(RobolectricTestRunner.class)
+public final class ClientInboundTest {
+
+ private static final Metadata.Key In this single-threaded test model, tasks posted to the main looper or transport executors
+ * are executed synchronously when the looper is idled.
+ */
+ private void drainExecutors() {
+ shadowOf(Looper.getMainLooper()).idle();
+ }
+
+ @Test
+ public void singleMessageEndToEndDelivery() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(1);
+
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("single-message-content"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages()).containsExactly("single-message-content");
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void multiMessageStreamingDelivery() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(3);
+
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("stream-message-1"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("stream-message-2"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("stream-message-3"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages())
+ .containsExactly("stream-message-1", "stream-message-2", "stream-message-3")
+ .inOrder();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void oversizedMessageLengthHeaderAbortsStream() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(1);
+
+ Parcel parcel = Parcel.obtain();
+ try {
+ parcel.writeInt(0); // placeholder for flags
+ parcel.writeInt(nextTxIndex++);
+ parcel.writeInt(1000); // claim message length 1000, but write nothing else
+ TransactionUtils.fillInFlags(parcel, FLAG_MESSAGE_DATA);
+ parcel.setDataPosition(0);
+ inbound.handleTransaction(parcel);
+ } finally {
+ parcel.recycle();
+ }
+
+ drainExecutors();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).hasCode(Status.Code.INTERNAL);
+ assertThat(listener.getClosedStatus().getDescription())
+ .contains("Message size is larger than remaining parcel size");
+ assertThat(listener.getReadMessages()).isEmpty();
+ }
+
+ @Test
+ public void zeroByteMessagePayload() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(1);
+
+ newStreamTxnToClientBuilder(nextTxIndex++).withMessage(utf8("")).dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages()).containsExactly("");
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void partialConsumptionFromMessageProducer() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(2);
+ listener.setReadPermits(1);
+
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("partial-consume-1"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("partial-consume-2"))
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ // Only partial-consume-1 was read by the listener even though 2 were requested
+ assertThat(listener.getReadMessages()).containsExactly("partial-consume-1");
+
+ MessageProducer producer = listener.pollMessageProducer();
+ assertThat(producer).isNotNull();
+
+ // Explicitly consume the second message from the producer outside the callback
+ String unconsumedMessage = FakeStreamListener.readString(producer.next());
+ assertThat(unconsumedMessage).isEqualTo("partial-consume-2");
+
+ // Producer is now drained
+ assertThat(producer.next()).isNull();
+ assertThat(listener.pollMessageProducer()).isNull();
+
+ // Listener read messages remain strictly unchanged
+ assertThat(listener.getReadMessages()).containsExactly("partial-consume-1");
+ }
+
+ @Test
+ public void deferredReadFlowControlMultipleMessagesAndSuffix() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+
+ // Messages and suffix arrive while 0 messages are requested.
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("deferred-msg-1"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("deferred-msg-2"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("deferred-msg-3"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+
+ // Initially 0 requested messages: nothing delivered, suffix not delivered.
+ drainExecutors();
+ assertThat(listener.getReadMessages()).isEmpty();
+ assertThat(listener.isClosed()).isFalse();
+
+ // Request 1 message: deferred-msg-1 delivered, suffix still not delivered.
+ clientStream.request(1);
+ drainExecutors();
+ assertThat(listener.getReadMessages()).containsExactly("deferred-msg-1");
+ assertThat(listener.isClosed()).isFalse();
+
+ // Request 1 message: deferred-msg-2 delivered, suffix still not delivered.
+ clientStream.request(1);
+ drainExecutors();
+ assertThat(listener.getReadMessages())
+ .containsExactly("deferred-msg-1", "deferred-msg-2")
+ .inOrder();
+ assertThat(listener.isClosed()).isFalse();
+
+ // Request 1 message: deferred-msg-3 delivered, all messages consumed, suffix is now delivered.
+ clientStream.request(1);
+ drainExecutors();
+ assertThat(listener.getReadMessages())
+ .containsExactly("deferred-msg-1", "deferred-msg-2", "deferred-msg-3")
+ .inOrder();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void multiPacketBlockFragmentationAndReassembly() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(1);
+
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessageFragment(utf8("first"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessageFragment(utf8("second"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withFinalMessageFragment(utf8("third"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages()).containsExactly("firstsecondthird");
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void outOfOrderMessageDeliveredBeforePrefix() throws Exception {
+ // Deliver message (Tx 1) BEFORE prefix (Tx 0)
+ newStreamTxnToClientBuilder(1).withMessage(utf8("some message")).dispatchTo(inbound);
+
+ // Deliver prefix (Tx 0)
+ newStreamTxnToClientBuilder(0).withPrefix(new Metadata()).dispatchTo(inbound);
+
+ // Request message after prefix has arrived
+ clientStream.request(1);
+
+ // Deliver suffix (Tx 2)
+ newStreamTxnToClientBuilder(2).withSuffix(Status.OK, new Metadata()).dispatchTo(inbound);
+
+ drainExecutors();
+ // Verify message and suffix are delivered
+ assertThat(listener.getReadMessages()).containsExactly("some message");
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void sequenceGapCausesBufferingUntilMissingTransactionArrives() throws Exception {
+ newStreamTxnToClientBuilder(0).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(2);
+
+ // Send index 2, skipping expected index 1
+ newStreamTxnToClientBuilder(2).withMessage(utf8("gap-message-2")).dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages()).isEmpty();
+
+ // Now send index 1
+ newStreamTxnToClientBuilder(1).withMessage(utf8("gap-message-1")).dispatchTo(inbound);
+
+ drainExecutors();
+ // Both messages delivered in order
+ assertThat(listener.getReadMessages())
+ .containsExactly("gap-message-1", "gap-message-2")
+ .inOrder();
+ }
+
+ @Test
+ public void outOfBandClose() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+
+ newOutOfBandCloseTxnBuilder(Status.CANCELLED.withDescription("remote cancelled RPC"))
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).hasCode(Status.Code.CANCELLED);
+ assertThat(listener.getClosedStatus().getDescription()).contains("remote cancelled RPC");
+ }
+
+ @Test
+ public void cleanupUnconsumedResourcesOnAbnormalClose() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ // Do not request messages so messages remain unconsumed in Inbound
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("unconsumed-1"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessage(utf8("unconsumed-2"))
+ .dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withMessageFragment(utf8("unconsumed-partial-block"))
+ .dispatchTo(inbound);
+
+ // Abort abnormally via out-of-band close
+ newOutOfBandCloseTxnBuilder(Status.UNAVAILABLE.withDescription("aborted")).dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(transport.getOngoingCalls()).isEmpty();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).hasCode(Status.Code.UNAVAILABLE);
+ assertThat(listener.getClosedStatus().getDescription()).contains("aborted");
+ assertThat(listener.getReadMessages()).isEmpty();
+ }
+
+ @Test
+ public void transactionsIgnoredAfterClosed() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ newOutOfBandCloseTxnBuilder(Status.CANCELLED).dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).hasCode(Status.Code.CANCELLED);
+
+ // Any subsequent transaction should be ignored silently
+ newStreamTxnToClientBuilder(nextTxIndex++).withMessage(utf8("ignored")).dispatchTo(inbound);
+ drainExecutors();
+ assertThat(listener.getReadMessages()).isEmpty();
+ }
+
+ @Test
+ public void allInOneUnaryTransaction() throws Exception {
+ newStreamTxnToClientBuilder(0)
+ .withPrefix(new Metadata())
+ .withMessage(utf8("all-in-one-message"))
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+ clientStream.request(1);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages()).containsExactly("all-in-one-message");
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ assertThat(listener.getClosedTrailers().keys()).isEmpty();
+ }
+
+ @Test
+ public void suffixWithNonOkStatusAndTrailers() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ Metadata trailers = new Metadata();
+ trailers.put(SOME_METADATA_KEY, "trailer-val");
+
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.NOT_FOUND.withDescription("item not found"), trailers)
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).hasCode(Status.Code.NOT_FOUND);
+ assertThat(listener.getClosedStatus().getDescription()).isEqualTo("item not found");
+ assertThat(listener.getClosedTrailers().get(SOME_METADATA_KEY)).isEqualTo("trailer-val");
+ }
+
+ @Test
+ public void prefixDeliversHeadersToListener() throws Exception {
+ Metadata headers = new Metadata();
+ headers.put(SOME_METADATA_KEY, "header-value");
+
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(headers).dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getHeaders().get(SOME_METADATA_KEY)).isEqualTo("header-value");
+ }
+
+ @Test
+ public void countsForInUse() {
+ assertThat(inbound.countsForInUse()).isTrue();
+
+ ClientStream balancerStream =
+ transport.newStream(
+ newDummyMethodDescriptorBuilder().build(),
+ new Metadata(),
+ CallOptions.DEFAULT.withOption(GrpcUtil.CALL_OPTIONS_RPC_OWNED_BY_BALANCER, true),
+ new ClientStreamTracer[0]);
+ balancerStream.start(mock(ClientStreamListener.class));
+ Inbound.ClientInbound notInUseInbound =
+ (Inbound.ClientInbound) transport.getOngoingCalls().get(inbound.callId + 1);
+ assertThat(notInUseInbound.countsForInUse()).isFalse();
+ }
+
+ @Test
+ public void excessRequestedMessagesDeliverCleanlyOnSuffix() throws Exception {
+ newStreamTxnToClientBuilder(nextTxIndex++).withPrefix(new Metadata()).dispatchTo(inbound);
+ clientStream.request(5);
+
+ newStreamTxnToClientBuilder(nextTxIndex++).withMessage(utf8("msg1")).dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++).withMessage(utf8("msg2")).dispatchTo(inbound);
+ newStreamTxnToClientBuilder(nextTxIndex++)
+ .withSuffix(Status.OK, new Metadata())
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(listener.getReadMessages()).containsExactly("msg1", "msg2").inOrder();
+ assertThat(listener.isClosed()).isTrue();
+ assertThat(listener.getClosedStatus()).isOk();
+ }
+
+ @Test
+ public void clientCancelUnregisters() throws Exception {
+ newStreamTxnToClientBuilder(0).withPrefix(new Metadata()).dispatchTo(inbound);
+ newStreamTxnToClientBuilder(1)
+ .withMessageFragment(utf8("unconsumed-partial"))
+ .dispatchTo(inbound);
+
+ clientStream.cancel(Status.CANCELLED.withDescription("client cancel"));
+
+ drainExecutors();
+ assertThat(transport.getOngoingCalls()).doesNotContainKey(inbound.callId);
+ }
+
+ @Test
+ public void abnormalCloseBeforeStreamStartDoesNotThrow() throws Exception {
+ transport.newStream(
+ newDummyMethodDescriptorBuilder().build(),
+ new Metadata(),
+ CallOptions.DEFAULT,
+ new ClientStreamTracer[0]);
+ Inbound.ClientInbound unstartedInbound =
+ (Inbound.ClientInbound) transport.getOngoingCalls().get(inbound.callId + 1);
+ assertThat(transport.getOngoingCalls()).containsKey(unstartedInbound.callId);
+
+ newOutOfBandCloseTxnBuilder(Status.CANCELLED.withDescription("remote abort"))
+ .dispatchTo(unstartedInbound);
+
+ drainExecutors();
+ assertThat(transport.getOngoingCalls()).doesNotContainKey(unstartedInbound.callId);
+ }
+
+ private static final class DummyClientTransportBuilder {
+ private final BinderClientTransportFactory.Builder factoryBuilder;
+
+ DummyClientTransportBuilder() {
+ MainThreadScheduledExecutorService mainThreadExecutor =
+ new MainThreadScheduledExecutorService();
+ factoryBuilder =
+ new BinderClientTransportFactory.Builder()
+ .setSourceContext(ApplicationProvider.getApplicationContext())
+ .setOffloadExecutorPool(new FixedObjectPool<>(mainThreadExecutor))
+ .setScheduledExecutorPool(new FixedObjectPool<>(mainThreadExecutor));
+ }
+
+ public BinderClientTransport build() {
+ SocketAddress serverAddress =
+ AndroidComponentAddress.forComponent(new ComponentName("fake.pkg", "fake.cls"));
+ BinderClientTransport transport =
+ new BinderClientTransportBuilder()
+ .setFactory(factoryBuilder.buildClientTransportFactory())
+ .setServerAddress(serverAddress)
+ .build();
+ // This hack lets us create a transport without the need for a real server for handshaking.
+ synchronized (transport) {
+ // Blackhole Outbound.
+ transport.setOutgoingBinder(mock(OneWayBinderProxy.class));
+ }
+ Runnable unused = transport.start(mock(ManagedClientTransport.Listener.class));
+ synchronized (transport) {
+ transport.setState(BinderTransport.TransportState.READY);
+ }
+ return transport;
+ }
+ }
+}
diff --git a/binder/src/test/java/io/grpc/binder/internal/ServerInboundTest.java b/binder/src/test/java/io/grpc/binder/internal/ServerInboundTest.java
new file mode 100644
index 00000000000..2677274396c
--- /dev/null
+++ b/binder/src/test/java/io/grpc/binder/internal/ServerInboundTest.java
@@ -0,0 +1,295 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.binder.internal;
+
+import static com.google.common.truth.Truth.assertThat;
+import static io.grpc.StatusSubject.assertThat;
+import static io.grpc.binder.internal.BinderTransport.FIRST_CALL_ID;
+import static io.grpc.binder.internal.TransactionBuilder.newOutOfBandCloseTxnBuilder;
+import static io.grpc.binder.internal.TransactionBuilder.newStreamTxnToServerBuilder;
+import static io.grpc.binder.internal.TransactionBuilder.utf8;
+import static java.util.Objects.requireNonNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.robolectric.Shadows.shadowOf;
+
+import android.os.IBinder;
+import android.os.Looper;
+import io.grpc.Metadata;
+import io.grpc.Status;
+import io.grpc.binder.internal.FakeServerTransportListener.CreatedStream;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.robolectric.RobolectricTestRunner;
+
+/**
+ * Unit tests for {@link Inbound.ServerInbound}.
+ *
+ * Focuses on server-specific inbound behaviors like method name extraction, initial headers and
+ * half-close handling. Inbound functions common to both client and server are tested in {@link
+ * ClientInboundTest}.
+ *
+ * Threading model: All Executors lead to Robolectric's main thread, where everything runs,
+ * including the test cases themselves. This makes it easy to write deterministic positive and
+ * negative assertions about listener callbacks because we can drain all executors and know that if
+ * the SUT was going to do something, it would have already happened. It certainly isn't realistic
+ * with respect to concurrency but that aspect is integration tested elsewhere (at a higher level).
+ */
+@RunWith(RobolectricTestRunner.class)
+public final class ServerInboundTest {
+
+ private static final Metadata.Key In this single-threaded test model, tasks posted to the main looper or transport executors
+ * are executed synchronously when the looper is idled.
+ */
+ private void drainExecutors() {
+ shadowOf(Looper.getMainLooper()).idle();
+ }
+
+ private static Inbound.ServerInbound getInboundOrDie(BinderTransport transport, int callId) {
+ return (Inbound.ServerInbound) requireNonNull(transport.getOngoingCalls().get(callId));
+ }
+
+ @Test
+ public void prefixInitializesServerStreamWithMethodAndHeaders() throws Exception {
+ Metadata headers = new Metadata();
+ headers.put(SOME_METADATA_KEY, "server-val");
+
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("my.custom.package.Service/StreamingCall", headers)
+ .dispatchTo(transport, FIRST_CALL_ID);
+
+ drainExecutors();
+ createdStream = transportListener.getOnlyCreatedStream();
+ assertThat(createdStream.getMethodName())
+ .isEqualTo("my.custom.package.Service/StreamingCall");
+ assertThat(createdStream.getHeaders().get(SOME_METADATA_KEY)).isEqualTo("server-val");
+ assertThat(createdStream.getStream()).isInstanceOf(MultiMessageServerStream.class);
+ assertThat(transport.getOngoingCalls()).containsKey(FIRST_CALL_ID);
+ }
+
+ @Test
+ public void singleMessageStreamTypeDetectedFromFlag() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("service/UnaryCall", new Metadata())
+ .withExpectSingleMessage()
+ .dispatchTo(transport, FIRST_CALL_ID);
+
+ drainExecutors();
+ createdStream = transportListener.getOnlyCreatedStream();
+ assertThat(createdStream.getMethodName()).isEqualTo("service/UnaryCall");
+ assertThat(createdStream.getStream()).isInstanceOf(SingleMessageServerStream.class);
+ }
+
+ @Test
+ public void singleMessageEndToEndDelivery() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("package.Service/Method", new Metadata())
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+ createdStream.getStream().request(1);
+
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("server-request-content"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++).withSuffix().dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages()).containsExactly("server-request-content");
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isTrue();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isFalse();
+ }
+
+ @Test
+ public void multiMessageStreamingDelivery() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("package.Service/Method", new Metadata())
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+ createdStream.getStream().request(3);
+
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("server-request-1"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("server-request-2"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("server-request-3"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++).withSuffix().dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages())
+ .containsExactly("server-request-1", "server-request-2", "server-request-3")
+ .inOrder();
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isTrue();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isFalse();
+ }
+
+ @Test
+ public void deferredReadFlowControlMultipleMessagesAndSuffix() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("package.Service/Method", new Metadata())
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+
+ // Messages and suffix arrive while 0 messages are requested.
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("deferred-msg-1"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("deferred-msg-2"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withMessage(utf8("deferred-msg-3"))
+ .dispatchTo(inbound);
+ newStreamTxnToServerBuilder(nextTxIndex++).withSuffix().dispatchTo(inbound);
+
+ // Initially 0 requested messages: nothing delivered, half-close not delivered.
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages()).isEmpty();
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isFalse();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+
+ // Request 1 message: deferred-msg-1 delivered, half-close still not delivered.
+ createdStream.getStream().request(1);
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages()).containsExactly("deferred-msg-1");
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isFalse();
+
+ // Request 1 message: deferred-msg-2 delivered, half-close still not delivered.
+ createdStream.getStream().request(1);
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages())
+ .containsExactly("deferred-msg-1", "deferred-msg-2")
+ .inOrder();
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isFalse();
+
+ // Request 1 message: deferred-msg-3 delivered, all messages consumed, half-close is now delivered.
+ createdStream.getStream().request(1);
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages())
+ .containsExactly("deferred-msg-1", "deferred-msg-2", "deferred-msg-3")
+ .inOrder();
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isTrue();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isFalse();
+ }
+
+ @Test
+ public void allInOneUnaryTransactionDeliversMessageAndHalfClose() throws Exception {
+ newStreamTxnToServerBuilder(0)
+ .withPrefix("package.Service/Method", new Metadata())
+ .withMessage(utf8("all-in-one-message"))
+ .withSuffix()
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+ createdStream.getStream().request(1);
+
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().getReadMessages()).containsExactly("all-in-one-message");
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isTrue();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isFalse();
+ }
+
+ @Test
+ public void clientHalfCloseDeliveredToListener() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("package.Service/Method", new Metadata())
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isFalse();
+
+ newStreamTxnToServerBuilder(nextTxIndex++).withSuffix().dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(createdStream.getStreamListener().isHalfClosed()).isTrue();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isFalse();
+ }
+
+ @Test
+ public void onCloseSentClosesStream() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("package.Service/Method", new Metadata())
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+ drainExecutors();
+ assertThat(transport.getOngoingCalls()).containsKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isFalse();
+
+ createdStream.getStream().close(Status.OK, new Metadata());
+ drainExecutors();
+ assertThat(transport.getOngoingCalls()).doesNotContainKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isTrue();
+ }
+
+ @Test
+ public void outOfBandCloseAbortsServerStream() throws Exception {
+ newStreamTxnToServerBuilder(nextTxIndex++)
+ .withPrefix("package.Service/Method", new Metadata())
+ .dispatchTo(transport, FIRST_CALL_ID);
+ Inbound.ServerInbound inbound = getInboundOrDie(transport, FIRST_CALL_ID);
+ createdStream = transportListener.getOnlyCreatedStream();
+
+ newOutOfBandCloseTxnBuilder(Status.CANCELLED.withDescription("client cancel"))
+ .dispatchTo(inbound);
+
+ drainExecutors();
+ assertThat(transport.getOngoingCalls()).doesNotContainKey(inbound.callId);
+ assertThat(createdStream.getStreamListener().isClosed()).isTrue();
+ assertThat(createdStream.getStreamListener().getClosedStatus()).hasCode(Status.Code.CANCELLED);
+ assertThat(createdStream.getStreamListener().getClosedStatus().getDescription()).contains("client cancel");
+ }
+}
diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/BinderServerTransportBuilder.java b/binder/src/testFixtures/java/io/grpc/binder/internal/BinderServerTransportBuilder.java
new file mode 100644
index 00000000000..7981d5be4e7
--- /dev/null
+++ b/binder/src/testFixtures/java/io/grpc/binder/internal/BinderServerTransportBuilder.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright 2026 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.binder.internal;
+
+import android.os.IBinder;
+import com.google.common.collect.ImmutableList;
+import io.grpc.Attributes;
+import io.grpc.ServerStreamTracer;
+import io.grpc.internal.FixedObjectPool;
+import io.grpc.internal.ObjectPool;
+import java.util.List;
+import java.util.concurrent.ScheduledExecutorService;
+
+/** Helps create {@link BinderServerTransport} instances without mentioning irrelevant details. */
+public class BinderServerTransportBuilder {
+ private ObjectPool