From f4937a044d98625189d47311fbf211eed72995ff Mon Sep 17 00:00:00 2001 From: jdcormie Date: Fri, 4 Sep 2026 18:56:39 -0700 Subject: [PATCH 1/3] binder: Fix OneWayBinderProxy to be a complete abstraction, removing getDelegate() Permits fake/mock implementations in tests that aren't actually backed by an IBinder at all. TAG=agy CONV=a7051e19-0fc1-42a4-8b24-5c2e3373aa0c --- .../grpc/binder/internal/BinderTransport.java | 4 +- .../binder/internal/OneWayBinderProxy.java | 82 ++++++++++++------- .../binder/internal/OneWayBinderProxies.java | 50 +++++++---- 3 files changed, 88 insertions(+), 48 deletions(-) diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java index c70cadc8749..cfb55f5f72a 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderTransport.java @@ -293,7 +293,7 @@ protected boolean setOutgoingBinder(OneWayBinderProxy binder) { binder = binderDecorator.decorate(binder); this.outgoingBinder = binder; try { - binder.getDelegate().linkToDeath(this, 0); + binder.linkToDeath(this, 0); return true; } catch (RemoteException re) { return false; @@ -367,7 +367,7 @@ final void sendSetupTransaction(OneWayBinderProxy iBinder) { private final void sendShutdownTransaction() { if (outgoingBinder != null) { try { - outgoingBinder.getDelegate().unlinkToDeath(this, 0); + outgoingBinder.unlinkToDeath(this, 0); } catch (NoSuchElementException e) { // Ignore. } diff --git a/binder/src/main/java/io/grpc/binder/internal/OneWayBinderProxy.java b/binder/src/main/java/io/grpc/binder/internal/OneWayBinderProxy.java index fd883ca3b62..4da63a07684 100644 --- a/binder/src/main/java/io/grpc/binder/internal/OneWayBinderProxy.java +++ b/binder/src/main/java/io/grpc/binder/internal/OneWayBinderProxy.java @@ -1,5 +1,7 @@ package io.grpc.binder.internal; +import static com.google.common.base.Preconditions.checkNotNull; + import android.os.Binder; import android.os.IBinder; import android.os.Parcel; @@ -10,7 +12,7 @@ import java.util.logging.Logger; /** - * Wraps an {@link IBinder} with a safe and uniformly asynchronous transaction API. + * A safe and uniformly asynchronous sink for "oneway" Binder transactions. * *

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/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java b/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java index c7eee06e73a..64dd7423237 100644 --- a/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/OneWayBinderProxies.java @@ -15,6 +15,7 @@ */ package io.grpc.binder.internal; +import android.os.IBinder; import android.os.RemoteException; import java.util.concurrent.BlockingQueue; import java.util.concurrent.LinkedBlockingQueue; @@ -23,6 +24,30 @@ /** A collection of {@link OneWayBinderProxy}-related test helpers. */ public final class OneWayBinderProxies { + /** Base class for {@link OneWayBinderProxy} decorators that forward all calls by default. */ + public abstract static class ForwardingOneWayBinderProxy extends OneWayBinderProxy { + protected final OneWayBinderProxy delegate; + + protected ForwardingOneWayBinderProxy(OneWayBinderProxy delegate) { + this.delegate = delegate; + } + + @Override + public void transact(int code, ParcelHolder data) throws RemoteException { + delegate.transact(code, data); + } + + @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); + } + } + /** * A {@link OneWayBinderProxy.Decorator} that blocks calling threads while an (external) test * provides the actual decoration. @@ -73,13 +98,11 @@ public OneWayBinderProxy decorate(OneWayBinderProxy in) { } /** A {@link OneWayBinderProxy} decorator whose transact method can artificially throw. */ - public static final class ThrowingOneWayBinderProxy extends OneWayBinderProxy { - private final OneWayBinderProxy wrapped; + public static final class ThrowingOneWayBinderProxy extends ForwardingOneWayBinderProxy { @Nullable private RemoteException remoteException; ThrowingOneWayBinderProxy(OneWayBinderProxy wrapped) { - super(wrapped.getDelegate()); - this.wrapped = wrapped; + super(wrapped); } /** @@ -97,21 +120,18 @@ public void transact(int code, ParcelHolder data) throws RemoteException { if (remoteException != null) { throw remoteException; } - wrapped.transact(code, data); + super.transact(code, data); } } /** * A {@link OneWayBinderProxy} decorator whose transact method can be configured to silently drop. */ - public static final class BlackHoleOneWayBinderProxy extends OneWayBinderProxy { - - private final OneWayBinderProxy wrapped; + public static final class BlackHoleOneWayBinderProxy extends ForwardingOneWayBinderProxy { private boolean dropAllTransactions; BlackHoleOneWayBinderProxy(OneWayBinderProxy wrapped) { - super(wrapped.getDelegate()); - this.wrapped = wrapped; + super(wrapped); } /** @@ -127,13 +147,13 @@ public void dropAllTransactions(boolean dropAllTransactions) { @Override public void transact(int code, ParcelHolder data) throws RemoteException { if (!dropAllTransactions) { - wrapped.transact(code, data); + super.transact(code, data); } } } /** A {@link OneWayBinderProxy} that queues transactions for a test to deliver manually later. */ - public static final class QueueingOneWayBinderProxy extends OneWayBinderProxy { + public static final class QueueingOneWayBinderProxy extends ForwardingOneWayBinderProxy { public static final class Transaction { public final int code; private final ParcelHolder parcel; @@ -145,11 +165,9 @@ public Transaction(int code, ParcelHolder parcel) { } private final BlockingQueue queue = new LinkedBlockingQueue<>(); - private final OneWayBinderProxy wrapped; public QueueingOneWayBinderProxy(OneWayBinderProxy wrapped) { - super(wrapped.getDelegate()); - this.wrapped = wrapped; + super(wrapped); } @Override @@ -171,7 +189,7 @@ public Transaction pollNextTransaction(long timeout, TimeUnit unit) * @throws IllegalStateException if transaction was already delivered once before */ public void deliver(Transaction transaction) throws RemoteException { - wrapped.transact(transaction.code, transaction.parcel); + delegate.transact(transaction.code, transaction.parcel); } } From e36a94d7880cc8d75b266678f20f777236c58785 Mon Sep 17 00:00:00 2001 From: jdcormie Date: Fri, 4 Sep 2026 18:57:18 -0700 Subject: [PATCH 2/3] binder: Add unit tests for Inbound behaviors not already tested at higher layer It's hard to write a true unit test for Inbound because of its many concrete dependencies. However, today it has *no* unit tests, so we can't safely refactor it. To get out of this fix, we introduce ClientInboundTest and ServerInboundTest that target Inbound's logic by way of the Stream and BinderTransport. TAG=agy CONV=a7051e19-0fc1-42a4-8b24-5c2e3373aa0c --- .../java/io/grpc/StatusSubject.java | 5 + .../internal/BinderServerTransportTest.java | 44 -- .../binder/internal/ClientInboundTest.java | 509 ++++++++++++++++++ .../binder/internal/ServerInboundTest.java | 295 ++++++++++ .../BinderServerTransportBuilder.java | 69 +++ .../internal/FakeClientStreamListener.java | 65 +++ .../internal/FakeServerStreamListener.java | 46 ++ .../internal/FakeServerTransportListener.java | 107 ++++ .../binder/internal/FakeStreamListener.java | 129 +++++ .../binder/internal/TransactionBuilder.java | 276 ++++++++++ 10 files changed, 1501 insertions(+), 44 deletions(-) create mode 100644 binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java create mode 100644 binder/src/test/java/io/grpc/binder/internal/ServerInboundTest.java create mode 100644 binder/src/testFixtures/java/io/grpc/binder/internal/BinderServerTransportBuilder.java create mode 100644 binder/src/testFixtures/java/io/grpc/binder/internal/FakeClientStreamListener.java create mode 100644 binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerStreamListener.java create mode 100644 binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerTransportListener.java create mode 100644 binder/src/testFixtures/java/io/grpc/binder/internal/FakeStreamListener.java create mode 100644 binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java diff --git a/api/src/testFixtures/java/io/grpc/StatusSubject.java b/api/src/testFixtures/java/io/grpc/StatusSubject.java index 0b00df96140..5975db604c0 100644 --- a/api/src/testFixtures/java/io/grpc/StatusSubject.java +++ b/api/src/testFixtures/java/io/grpc/StatusSubject.java @@ -17,6 +17,7 @@ package io.grpc; import static com.google.common.truth.Fact.fact; +import static com.google.common.truth.Truth.assertAbout; import com.google.common.truth.FailureMetadata; import com.google.common.truth.Subject; @@ -31,6 +32,10 @@ public static Subject.Factory status() { return statusFactory; } + public static StatusSubject assertThat(@Nullable Status status) { + return assertAbout(status()).that(status); + } + private final Status actual; private StatusSubject(FailureMetadata metadata, @Nullable Status subject) { 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 executorServicePool; - Attributes attributes; - List streamTracerFactories; - OneWayBinderProxy.Decorator binderDecorator; - IBinder callbackBinder; - - public BinderServerTransport build() { - return BinderServerTransport.create( - executorServicePool, attributes, streamTracerFactories, binderDecorator, callbackBinder); - } - - public BinderServerTransportBuilder setExecutorServicePool( - ObjectPool executorServicePool) { - this.executorServicePool = executorServicePool; - return this; - } - - public BinderServerTransportBuilder setAttributes(Attributes attributes) { - this.attributes = attributes; - return this; - } - - public BinderServerTransportBuilder setStreamTracerFactories( - List streamTracerFactories) { - this.streamTracerFactories = streamTracerFactories; - return this; - } - - public BinderServerTransportBuilder setBinderDecorator( - OneWayBinderProxy.Decorator binderDecorator) { - this.binderDecorator = binderDecorator; - return this; - } - - public BinderServerTransportBuilder setCallbackBinder(IBinder callbackBinder) { - this.callbackBinder = callbackBinder; - return this; - } - } } diff --git a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java new file mode 100644 index 00000000000..791d9a86473 --- /dev/null +++ b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java @@ -0,0 +1,509 @@ +/* + * 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.TransactionBuilder.newOutOfBandCloseTxnBuilder; +import static io.grpc.binder.internal.TransactionBuilder.newStreamTxnToClientBuilder; +import static io.grpc.binder.internal.TransactionBuilder.utf8; +import static io.grpc.binder.internal.TransactionUtils.FLAG_MESSAGE_DATA; +import static org.mockito.Mockito.mock; +import static org.robolectric.Shadows.shadowOf; + +import android.content.ComponentName; +import android.os.Looper; +import android.os.Parcel; +import androidx.test.core.app.ApplicationProvider; +import io.grpc.CallOptions; +import io.grpc.ClientStreamTracer; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.Status; +import io.grpc.StringMarshaller; +import io.grpc.binder.AndroidComponentAddress; +import io.grpc.internal.ClientStream; +import io.grpc.internal.ClientStreamListener; +import io.grpc.internal.FixedObjectPool; +import io.grpc.internal.GrpcUtil; +import io.grpc.internal.ManagedClientTransport; +import io.grpc.internal.StreamListener.MessageProducer; +import java.net.SocketAddress; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.robolectric.RobolectricTestRunner; + +/** + * Unit tests for {@link Inbound.ClientInbound}. + * + *

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 SOME_METADATA_KEY = + Metadata.Key.of("some-metadata-key", Metadata.ASCII_STRING_MARSHALLER); + + private static final MethodDescriptor methodDescriptor = + MethodDescriptor.newBuilder() + .setType(MethodDescriptor.MethodType.UNKNOWN) + .setFullMethodName("package.Service/Method") + .setRequestMarshaller(StringMarshaller.INSTANCE) + .setResponseMarshaller(StringMarshaller.INSTANCE) + .build(); + + private BinderClientTransport transport; + private Inbound.ClientInbound inbound; + private int nextTxIndex; // Only used in test cases where the index value is unimportant. + private ClientStream clientStream; + private FakeClientStreamListener listener; + + @Before + public void setUp() throws Exception { + // Inbound is presently impossible to create in isolation. We need a dummy instance of the + // transport to own new Inbounds and provide its deps. TODO(jdcormie): Refactor Inbound so it + // can be unit tested without hacks. + transport = createDummyTransport(); + listener = new FakeClientStreamListener(); + clientStream = + transport.newStream( + methodDescriptor, new Metadata(), CallOptions.DEFAULT, new ClientStreamTracer[0]); + clientStream.start(listener); + clientStream.writeMessage(methodDescriptor.getRequestMarshaller().stream("request")); + inbound = + (Inbound.ClientInbound) transport.getOngoingCalls().get(BinderTransport.FIRST_CALL_ID); + } + + /** + * Drains any pending asynchronous tasks on transport executors before asserting state. + * + *

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 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( + methodDescriptor, + 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( + methodDescriptor, 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 BinderClientTransport createDummyTransport() { + MainThreadScheduledExecutorService mainThreadExecutor = + new MainThreadScheduledExecutorService(); + BinderClientTransportFactory factory = + new BinderClientTransportFactory.Builder() + .setSourceContext(ApplicationProvider.getApplicationContext()) + .setOffloadExecutorPool(new FixedObjectPool<>(mainThreadExecutor)) + .setScheduledExecutorPool(new FixedObjectPool<>(mainThreadExecutor)) + .buildClientTransportFactory(); + SocketAddress serverAddress = + AndroidComponentAddress.forComponent(new ComponentName("fake.pkg", "fake.cls")); + BinderClientTransport transport = + new BinderClientTransportBuilder() + .setFactory(factory) + .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 SOME_METADATA_KEY = + Metadata.Key.of("some-metadata-key", Metadata.ASCII_STRING_MARSHALLER); + + private BinderServerTransport transport; + private int nextTxIndex; // Only used in test cases where the index value is unimportant. + private FakeServerTransportListener transportListener; + private CreatedStream createdStream; + + @Before + public void setUp() throws Exception { + // ServerInbound is presently impossible to create in isolation. We need a dummy instance of the + // transport to own new Inbounds and provide its deps. TODO(jdcormie): Refactor Inbound so it + // can be unit tested without hacks. + IBinder mockBinder = mock(IBinder.class); // Black hole Outbound. + when(mockBinder.transact(anyInt(), any(), any(), anyInt())).thenReturn(true); + transport = new BinderServerTransportBuilder().setCallbackBinder(mockBinder).build(); + + transportListener = new FakeServerTransportListener<>(FakeServerStreamListener::new); + transport.start(transportListener); + } + + /** + * Drains any pending asynchronous tasks on transport executors before asserting state. + * + *

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 executorServicePool = + new FixedObjectPool<>(new MainThreadScheduledExecutorService()); + private Attributes attributes = Attributes.EMPTY; + private List streamTracerFactories = ImmutableList.of(); + private OneWayBinderProxy.Decorator binderDecorator = OneWayBinderProxy.IDENTITY_DECORATOR; + private IBinder callbackBinder; + + public BinderServerTransportBuilder setExecutorServicePool( + ObjectPool executorServicePool) { + this.executorServicePool = executorServicePool; + return this; + } + + public BinderServerTransportBuilder setAttributes(Attributes attributes) { + this.attributes = attributes; + return this; + } + + public BinderServerTransportBuilder setStreamTracerFactories( + List streamTracerFactories) { + this.streamTracerFactories = streamTracerFactories; + return this; + } + + public BinderServerTransportBuilder setBinderDecorator( + OneWayBinderProxy.Decorator binderDecorator) { + this.binderDecorator = binderDecorator; + return this; + } + + public BinderServerTransportBuilder setCallbackBinder(IBinder callbackBinder) { + this.callbackBinder = callbackBinder; + return this; + } + + public BinderServerTransport build() { + return BinderServerTransport.create( + executorServicePool, attributes, streamTracerFactories, binderDecorator, callbackBinder); + } +} diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/FakeClientStreamListener.java b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeClientStreamListener.java new file mode 100644 index 00000000000..10c6c90bee5 --- /dev/null +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeClientStreamListener.java @@ -0,0 +1,65 @@ +/* + * 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.base.Preconditions.checkState; + +import io.grpc.Metadata; +import io.grpc.Status; +import io.grpc.internal.ClientStreamListener; +import javax.annotation.Nullable; + +/** Fake {@link ClientStreamListener} for capturing headers, trailers, status, and messages. */ +public final class FakeClientStreamListener extends FakeStreamListener + implements ClientStreamListener { + @Nullable private Metadata headers; + @Nullable private RpcProgress closedProgress; + @Nullable private Metadata closedTrailers; + + @Override + public void headersRead(Metadata headers) { + checkState(!isClosed(), "headersRead invoked after closed"); + checkState(this.headers == null, "headersRead invoked more than once"); + this.headers = headers; + } + + @Override + public void closed(Status status, RpcProgress rpcProgress, Metadata trailers) { + checkState(!isClosed(), "closed invoked more than once"); + this.closedStatus = status; + this.closedProgress = rpcProgress; + this.closedTrailers = trailers; + } + + /** Returns the initial metadata headers received, or {@code null} if none. */ + @Nullable + public Metadata getHeaders() { + return headers; + } + + /** Returns the RPC progress passed to {@link #closed}, or {@code null} if not closed. */ + @Nullable + public RpcProgress getClosedProgress() { + return closedProgress; + } + + /** Returns the trailing metadata passed to {@link #closed}, or {@code null} if not closed. */ + @Nullable + public Metadata getClosedTrailers() { + return closedTrailers; + } +} diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerStreamListener.java b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerStreamListener.java new file mode 100644 index 00000000000..bf3265b046c --- /dev/null +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerStreamListener.java @@ -0,0 +1,46 @@ +/* + * 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.base.Preconditions.checkState; + +import io.grpc.Status; +import io.grpc.internal.ServerStreamListener; + +/** Fake {@link ServerStreamListener} for capturing half-close, status, and messages. */ +public final class FakeServerStreamListener extends FakeStreamListener + implements ServerStreamListener { + private boolean halfClosed; + + @Override + public void halfClosed() { + checkState(!isClosed(), "halfClosed invoked after closed"); + checkState(!halfClosed, "halfClosed invoked more than once"); + this.halfClosed = true; + } + + @Override + public void closed(Status status) { + checkState(!isClosed(), "closed invoked more than once"); + this.closedStatus = status; + } + + /** Returns whether {@link #halfClosed} was called. */ + public boolean isHalfClosed() { + return halfClosed; + } +} diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerTransportListener.java b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerTransportListener.java new file mode 100644 index 00000000000..37c69fde6e2 --- /dev/null +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeServerTransportListener.java @@ -0,0 +1,107 @@ +/* + * 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 java.util.Objects.requireNonNull; + +import com.google.common.collect.Iterables; +import io.grpc.Attributes; +import io.grpc.Metadata; +import io.grpc.internal.ServerStream; +import io.grpc.internal.ServerStreamListener; +import io.grpc.internal.ServerTransportListener; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.function.Supplier; + +/** + * Fake {@link ServerTransportListener} capturing inbound stream creations and attributes. + * + *

This class is not thread-safe. Tests must externally synchronize their assertions with + * callbacks to this listener from gRPC threads. + */ +public final class FakeServerTransportListener + implements ServerTransportListener { + + /** Encapsulates a recorded stream creation event on this transport. */ + public static final class CreatedStream { + private final ServerStream stream; + private final String methodName; + private final Metadata headers; + private final L streamListener; + + public CreatedStream( + ServerStream stream, + String methodName, + Metadata headers, + L streamListener) { + this.stream = requireNonNull(stream, "stream"); + this.methodName = requireNonNull(methodName, "methodName"); + this.headers = requireNonNull(headers, "headers"); + this.streamListener = requireNonNull(streamListener, "streamListener"); + } + + public ServerStream getStream() { + return stream; + } + + public String getMethodName() { + return methodName; + } + + public Metadata getHeaders() { + return headers; + } + + public L getStreamListener() { + return streamListener; + } + } + + private final List> createdStreams = new ArrayList<>(); + private final Supplier listenerFactory; + + public FakeServerTransportListener(Supplier listenerFactory) { + this.listenerFactory = requireNonNull(listenerFactory, "listenerFactory"); + } + + @Override + public void streamCreated(ServerStream stream, String methodName, Metadata headers) { + L streamListener = listenerFactory.get(); + // grpc-binder (incorrectly) assumes setListener() will be called before streamCreated() returns :( + stream.setListener(streamListener); + createdStreams.add(new CreatedStream<>(stream, methodName, headers, streamListener)); + } + + @Override + public Attributes transportReady(Attributes attributes) { + return attributes; + } + + @Override + public void transportTerminated() {} + + public List> getCreatedStreams() { + return Collections.unmodifiableList(createdStreams); + } + + public CreatedStream getOnlyCreatedStream() { + return Iterables.getOnlyElement(createdStreams); + } +} + diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/FakeStreamListener.java b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeStreamListener.java new file mode 100644 index 00000000000..c2a2f5eb3f6 --- /dev/null +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/FakeStreamListener.java @@ -0,0 +1,129 @@ +/* + * 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.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; +import static java.nio.charset.StandardCharsets.UTF_8; + +import com.google.common.collect.ImmutableList; +import com.google.common.io.ByteStreams; +import io.grpc.Status; +import io.grpc.internal.StreamListener; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import javax.annotation.Nullable; + +/** + * Fake {@link StreamListener} that eagerly reads and records incoming stream messages. + * + *

During {@link #messagesAvailable(MessageProducer)}, the listener reads up to its remaining + * permit budget (default: unlimited) and records them in {@link #getReadMessages()}. If reading + * stops because permits ran out, the {@link MessageProducer} is saved for retrieval via {@link + * #pollMessageProducer()}. + * + *

This class is not thread-safe. Tests must synchronize their own state mutations and assertions + * with callbacks dispatched from gRPC threads. + */ +public class FakeStreamListener implements StreamListener { + private final List readMessages = new ArrayList<>(); + private final Queue messageProducers = new ArrayDeque<>(); + private int readPermitsRemaining = Integer.MAX_VALUE; + @Nullable protected Status closedStatus; + + /** + * Sets the exact number of messages the listener is permitted to read in subsequent {@link + * #messagesAvailable} callbacks. + */ + public void setReadPermits(int permits) { + checkArgument(permits >= 0, "permits must be non-negative"); + this.readPermitsRemaining = permits; + } + + /** Adds additional message read permits to the listener's budget. */ + public void addReadPermits(int permits) { + checkArgument(permits >= 0, "permits must be non-negative"); + checkState( + Integer.MAX_VALUE - this.readPermitsRemaining >= permits, "readPermitsRemaining overflow"); + this.readPermitsRemaining += permits; + } + + /** + * Polls and removes the next {@link MessageProducer} whose reading was halted for lack of + * permits, or {@code null} if none. + * + *

Note: The returned {@link MessageProducer} may be empty if the available permits were + * exactly enough to drain it. + */ + @Nullable + public MessageProducer pollMessageProducer() { + return messageProducers.poll(); + } + + @Override + public void messagesAvailable(MessageProducer producer) { + checkState(!isClosed(), "messagesAvailable invoked after closed"); + while (readPermitsRemaining > 0) { + InputStream stream = producer.next(); + if (stream == null) { + return; + } + readPermitsRemaining--; + try { + readMessages.add(readString(stream)); + } catch (IOException e) { + throw new AssertionError(e); + } + } + messageProducers.add(producer); + } + + /** Decodes the entire contents of {@code stream} as a UTF-8 string and closes the stream. */ + public static String readString(InputStream stream) throws IOException { + checkNotNull(stream, "stream"); + try (InputStream is = stream) { + return new String(ByteStreams.toByteArray(is), UTF_8); + } + } + + /** Returns an immutable snapshot of all messages read by the listener in order. */ + public ImmutableList getReadMessages() { + return ImmutableList.copyOf(readMessages); + } + + /** Returns the status passed to {@code closed()}, or {@code null} if not closed. */ + @Nullable + public Status getClosedStatus() { + return closedStatus; + } + + /** Returns whether the stream has been closed. */ + public boolean isClosed() { + return closedStatus != null; + } + + @Override + public void onReady() { + checkState(!isClosed(), "onReady invoked after closed"); + // Could maintain an onReady counter here if needed. + } +} diff --git a/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java b/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java new file mode 100644 index 00000000000..e5312f8234f --- /dev/null +++ b/binder/src/testFixtures/java/io/grpc/binder/internal/TransactionBuilder.java @@ -0,0 +1,276 @@ +/* + * 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 io.grpc.binder.internal.TransactionUtils.FLAG_MESSAGE_DATA; +import static io.grpc.binder.internal.TransactionUtils.FLAG_MESSAGE_DATA_IS_PARTIAL; +import static io.grpc.binder.internal.TransactionUtils.FLAG_OUT_OF_BAND_CLOSE; +import static io.grpc.binder.internal.TransactionUtils.FLAG_PREFIX; +import static io.grpc.binder.internal.TransactionUtils.FLAG_SUFFIX; +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Objects.requireNonNull; + +import android.os.Parcel; +import com.google.errorprone.annotations.CanIgnoreReturnValue; +import io.grpc.Metadata; +import io.grpc.Status; + +/** + * Builds and dispatches grpc-binder transactions for unit testing. + * + *

This class implements the [grpc-binder + * wireformat](https://github.com/grpc/proposal/blob/master/L73-java-binderchannel/wireformat.md). + * It makes low level unit tests easy to write and easy to read. It's intentionally difficult, but + * not impossible, to dispatch a transaction that violates the wireformat -- most mistakes will fail + * to compile. + */ +public abstract class TransactionBuilder { + + /** Creates a builder for a client-to-server (request stream) transaction. */ + public static ServerStreamTxnBuilder newStreamTxnToServerBuilder(int index) { + return new ServerStreamTxnBuilder(index); + } + + /** Creates a builder for a server-to-client (response stream) transaction. */ + public static ClientStreamTxnBuilder newStreamTxnToClientBuilder(int index) { + return new ClientStreamTxnBuilder(index); + } + + /** Creates a builder for an out-of-band close transaction. */ + public static OutOfBandCloseTxnBuilder newOutOfBandCloseTxnBuilder(Status status) { + return new OutOfBandCloseTxnBuilder(status); + } + + /** Functional interface for parcel consumers that may throw checked exceptions. */ + @FunctionalInterface + public interface ParcelConsumer { + void accept(Parcel parcel) throws Exception; + } + + /** Dispatches the synthesized parcel to a generic consumer, recycling the parcel afterwards. */ + public abstract void dispatchTo(ParcelConsumer consumer) throws Exception; + + /** Dispatches the synthesized parcel to a transport for the given callId. */ + public final void dispatchTo(BinderTransport transport, int callId) throws Exception { + dispatchTo(parcel -> transport.handleTransaction(callId, parcel)); + } + + /** Dispatches the synthesized parcel to an inbound handler. */ + public final void dispatchTo(Inbound inbound) throws Exception { + dispatchTo(inbound::handleTransaction); + } + + /** + * Base builder for in-band streaming transactions containing sequence index and message chunks. + */ + public abstract static class StreamTransactionBuilder> + extends TransactionBuilder { + protected final int index; + protected int flags; + protected byte[] messageData; + + protected StreamTransactionBuilder(int index) { + this.index = index; + } + + protected abstract B self(); + + protected abstract void writePrefix(Parcel parcel) throws Exception; + + protected abstract int writeSuffix(Parcel parcel) throws Exception; + + /** Appends complete message data from a byte array payload. */ + @CanIgnoreReturnValue + public final B withMessage(byte[] data) { + this.flags |= FLAG_MESSAGE_DATA; + this.messageData = requireNonNull(data, "data"); + return self(); + } + + /** Appends a partial message fragment from a byte array payload. */ + @CanIgnoreReturnValue + public final B withMessageFragment(byte[] data) { + this.flags |= FLAG_MESSAGE_DATA | FLAG_MESSAGE_DATA_IS_PARTIAL; + this.messageData = requireNonNull(data, "data"); + return self(); + } + + /** Appends the final message fragment from a byte array payload. */ + @CanIgnoreReturnValue + public final B withFinalMessageFragment(byte[] data) { + return withMessage(data); + } + + @Override + public final void dispatchTo(ParcelConsumer consumer) throws Exception { + Parcel parcel = Parcel.obtain(); + try { + parcel.writeInt(0); // placeholder for flags + parcel.writeInt(index); + writePrefix(parcel); + if ((flags & FLAG_MESSAGE_DATA) != 0) { + parcel.writeInt(messageData.length); + if (messageData.length > 0) { + parcel.writeByteArray(messageData); + } + } + int computedFlags = flags | writeSuffix(parcel); + TransactionUtils.fillInFlags(parcel, computedFlags); + parcel.setDataPosition(0); + consumer.accept(parcel); + } finally { + parcel.recycle(); + } + } + } + + /** Builder for client-to-server (request) stream transactions. */ + public static final class ServerStreamTxnBuilder + extends StreamTransactionBuilder { + private String methodName; + private Metadata headers; + + private ServerStreamTxnBuilder(int index) { + super(index); + } + + @Override + protected ServerStreamTxnBuilder self() { + return this; + } + + /** Sets the client prefix with the target RPC method name and initial request headers. */ + @CanIgnoreReturnValue + public ServerStreamTxnBuilder withPrefix(String methodName, Metadata headers) { + this.flags |= FLAG_PREFIX; + this.methodName = requireNonNull(methodName, "methodName"); + this.headers = requireNonNull(headers, "headers"); + return this; + } + + /** Sets the flag indicating that this RPC expects a single unary response. */ + @CanIgnoreReturnValue + public ServerStreamTxnBuilder withExpectSingleMessage() { + this.flags |= TransactionUtils.FLAG_EXPECT_SINGLE_MESSAGE; + return this; + } + + /** Sets the client half-close / end-of-stream suffix flag. */ + @CanIgnoreReturnValue + public ServerStreamTxnBuilder withSuffix() { + this.flags |= FLAG_SUFFIX; + return this; + } + + @Override + protected void writePrefix(Parcel parcel) throws Exception { + if ((flags & FLAG_PREFIX) != 0) { + parcel.writeString(methodName); + MetadataHelper.writeMetadata(parcel, headers); + } + } + + @Override + protected int writeSuffix(Parcel parcel) { + // Client-to-server suffix has no payload. + return 0; + } + } + + /** Builder for server-to-client (response) stream transactions. */ + public static final class ClientStreamTxnBuilder + extends StreamTransactionBuilder { + private Metadata headers; + private Status status; + private Metadata trailers; + + private ClientStreamTxnBuilder(int index) { + super(index); + } + + @Override + protected ClientStreamTxnBuilder self() { + return this; + } + + /** Sets the server prefix with initial response headers. */ + @CanIgnoreReturnValue + public ClientStreamTxnBuilder withPrefix(Metadata headers) { + this.flags |= FLAG_PREFIX; + this.headers = requireNonNull(headers, "headers"); + return this; + } + + /** Sets the server suffix with terminal status and trailing metadata. */ + @CanIgnoreReturnValue + public ClientStreamTxnBuilder withSuffix(Status status, Metadata trailers) { + this.flags |= FLAG_SUFFIX; + this.status = requireNonNull(status, "status"); + this.trailers = requireNonNull(trailers, "trailers"); + return this; + } + + @Override + protected void writePrefix(Parcel parcel) throws Exception { + if ((flags & FLAG_PREFIX) != 0) { + MetadataHelper.writeMetadata(parcel, headers); + } + } + + @Override + protected int writeSuffix(Parcel parcel) throws Exception { + if ((flags & FLAG_SUFFIX) != 0) { + int statusFlags = TransactionUtils.writeStatus(parcel, status); + MetadataHelper.writeMetadata(parcel, trailers); + return statusFlags; + } + return 0; + } + } + + /** Builder for out-of-band close transactions. */ + public static final class OutOfBandCloseTxnBuilder extends TransactionBuilder { + private final Status status; + + private OutOfBandCloseTxnBuilder(Status status) { + this.status = requireNonNull(status, "status"); + } + + @Override + public void dispatchTo(ParcelConsumer consumer) throws Exception { + Parcel parcel = Parcel.obtain(); + try { + parcel.writeInt(0); + int flags = FLAG_OUT_OF_BAND_CLOSE | TransactionUtils.writeStatus(parcel, status); + TransactionUtils.fillInFlags(parcel, flags); + parcel.setDataPosition(0); + consumer.accept(parcel); + } finally { + parcel.recycle(); + } + } + } + + /** + * Encodes the given string to bytes using UTF-8. + * + *

Convenient for unit tests that use string literals for payloads. + */ + public static byte[] utf8(String string) { + return string.getBytes(UTF_8); + } +} From ab3fd566964edd2659d519c5afeae9e4d2e3b152 Mon Sep 17 00:00:00 2001 From: jdcormie Date: Fri, 4 Sep 2026 18:57:41 -0700 Subject: [PATCH 3/3] binder: Look for complete messages after prefix-only transaction drains initial queued slot TAG=agy CONV=a7051e19-0fc1-42a4-8b24-5c2e3373aa0c --- .../java/io/grpc/binder/internal/Inbound.java | 1 + .../binder/internal/ClientInboundTest.java | 22 +++++++++++++++++++ 2 files changed, 23 insertions(+) diff --git a/binder/src/main/java/io/grpc/binder/internal/Inbound.java b/binder/src/main/java/io/grpc/binder/internal/Inbound.java index 83fc8273d6f..5b234149bf0 100644 --- a/binder/src/main/java/io/grpc/binder/internal/Inbound.java +++ b/binder/src/main/java/io/grpc/binder/internal/Inbound.java @@ -367,6 +367,7 @@ final synchronized void handleTransaction(Parcel parcel) { // The first transaction arrived, but it contained no message data. queuedTransactionData.remove(0); firstQueuedTransactionIndex += 1; + lookForCompleteMessage(); } } reportInboundSize(parcel.dataSize()); diff --git a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java index 791d9a86473..da979094d67 100644 --- a/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java +++ b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java @@ -300,6 +300,28 @@ public void multiPacketBlockFragmentationAndReassembly() throws Exception { 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);