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/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/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/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/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..da979094d67 --- /dev/null +++ b/binder/src/test/java/io/grpc/binder/internal/ClientInboundTest.java @@ -0,0 +1,531 @@ +/* + * 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 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( + 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/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); } } 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); + } +}