Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions api/src/main/java/io/grpc/ClientStreamTracer.java
Original file line number Diff line number Diff line change
Expand Up @@ -99,6 +99,15 @@ public void inboundTrailers(Metadata trailers) {
public void addOptionalLabel(String key, String value) {
}

/**
* The stream was cancelled from the client side before a normal response was received.
*
* @param status the cancellation status
* @since 1.84.0
*/
public void cancelled(Status status) {
}

/**
* Factory class for {@link ClientStreamTracer}.
*/
Expand Down
3 changes: 3 additions & 0 deletions binder/src/main/java/io/grpc/binder/internal/Inbound.java
Original file line number Diff line number Diff line change
Expand Up @@ -268,6 +268,9 @@ private final void deliverInternal() {

@GuardedBy("this")
final void closeOnCancel(Status status) {
if (!isClosed() && statsTraceContext != null) {
statsTraceContext.clientCancelled(status);
}
closeAbnormal(Status.CANCELLED, status, false);
}

Expand Down
16 changes: 13 additions & 3 deletions core/src/main/java/io/grpc/internal/AbstractClientStream.java
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,9 @@ public final void halfClose() {
@Override
public final void cancel(Status reason) {
Preconditions.checkArgument(!reason.isOk(), "Should not cancel with OK status");
if (cancelled || transportState().isListenerClosed()) {
return;
}
cancelled = true;
abstractClientStreamSink().cancel(reason);
}
Expand Down Expand Up @@ -251,6 +254,10 @@ protected TransportState(
}
}

protected final boolean isListenerClosed() {
return listenerClosed;
}

private void setFullStreamDecompression(boolean fullStreamDecompression) {
this.fullStreamDecompression = fullStreamDecompression;
}
Expand Down Expand Up @@ -435,13 +442,13 @@ public final void transportReportStatus(

if (deframerClosed) {
deframerClosedTask = null;
closeListener(status, rpcProgress, trailers);
closeListener(status, rpcProgress, trailers, stopDelivery);
} else {
deframerClosedTask =
new Runnable() {
@Override
public void run() {
closeListener(status, rpcProgress, trailers);
closeListener(status, rpcProgress, trailers, stopDelivery);
}
};
closeDeframer(stopDelivery);
Expand All @@ -454,9 +461,12 @@ public void run() {
* @throws IllegalStateException if the call has not yet been started.
*/
private void closeListener(
Status status, RpcProgress rpcProgress, Metadata trailers) {
Status status, RpcProgress rpcProgress, Metadata trailers, boolean stopDelivery) {
if (!listenerClosed) {
listenerClosed = true;
if (stopDelivery) {
statsTraceCtx.clientCancelled(status);
}
statsTraceCtx.streamClosed(status);
if (getTransportTracer() != null) {
getTransportTracer().reportStreamClosed(status.isOk());
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,11 @@ public void addOptionalLabel(String key, String value) {
delegate().addOptionalLabel(key, value);
}

@Override
public void cancelled(Status status) {
delegate().cancelled(status);
}

@Override
public void streamClosed(Status status) {
delegate().streamClosed(status);
Expand Down
13 changes: 13 additions & 0 deletions core/src/main/java/io/grpc/internal/StatsTraceContext.java
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,19 @@ public void serverCallMethodResolved(MethodDescriptor<?, ?> method) {
}
}

/**
* See {@link ClientStreamTracer#cancelled}. For client-side only.
*
* <p>Called from abstract stream implementations.
*/
public void clientCancelled(Status status) {
for (StreamTracer tracer : tracers) {
if (tracer instanceof ClientStreamTracer) {
((ClientStreamTracer) tracer).cancelled(status);
}
}
}

/**
* See {@link StreamTracer#streamClosed}. This may be called multiple times, and only the first
* value will be taken.
Expand Down
134 changes: 134 additions & 0 deletions core/src/test/java/io/grpc/internal/AbstractClientStreamTest.java
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

import io.grpc.Attributes;
import io.grpc.CallOptions;
import io.grpc.ClientStreamTracer;
import io.grpc.Codec;
import io.grpc.Deadline;
import io.grpc.Grpc;
Expand Down Expand Up @@ -155,6 +156,139 @@ public void cancel(Status errorStatus) {
verify(mockListener).closed(any(Status.class), same(PROCESSED), any(Metadata.class));
}

@Test
public void cancel_notifiesStatsTraceContext() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {
@Override
public void cancel(Status errorStatus) {
state.transportReportStatus(errorStatus, true, new Metadata());
}
}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

Status cancelStatus = Status.CANCELLED.withDescription("Cancelled by test");
stream.cancel(cancelStatus);

verify(mockTracer).cancelled(cancelStatus);
}

@Test
public void transportReportStatus_okFirst_lateCancellationDoesNotNotifyTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {
@Override
public void cancel(Status errorStatus) {
state.transportReportStatus(errorStatus, true, new Metadata());
}
}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Report Status.OK first
state.transportReportStatus(Status.OK, false, new Metadata());

// Subsequent late cancellation
verify(mockTracer, never()).cancelled(any(Status.class));
}

@Test
public void transportReportStatus_stopDeliveryFalse_doesNotNotifyTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {},
customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Server-initiated CANCELLED (stopDelivery = false)
state.transportReportStatus(Status.CANCELLED, false, new Metadata());

verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer).streamClosed(Status.CANCELLED);
}

@Test
public void transportReportStatus_stopDeliveryTrue_notifiesTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(allocator, state, new BaseSink() {},
customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Client/Transport-initiated cancellation (stopDelivery = true)
Status cancelStatus = Status.CANCELLED.withDescription("Client cancelled");
state.transportReportStatus(cancelStatus, true, new Metadata());

verify(mockTracer).cancelled(cancelStatus);
verify(mockTracer).streamClosed(cancelStatus);
}

@Test
public void transportReportStatus_stopDeliveryFalse_deadlineExceeded_noTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Server-initiated DEADLINE_EXCEEDED (stopDelivery = false)
Status status = Status.DEADLINE_EXCEEDED.withDescription("Server deadline exceeded");
state.transportReportStatus(status, false, new Metadata());

verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer).streamClosed(status);
}

@Test
public void transportReportStatus_stopDeliveryTrue_deadlineExceeded_notifiesTracerCancelled() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
final BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Client/Transport-initiated deadline exceeded (stopDelivery = true)
Status status = Status.DEADLINE_EXCEEDED.withDescription("Client deadline exceeded");
state.transportReportStatus(status, true, new Metadata());

verify(mockTracer).cancelled(status);
verify(mockTracer).streamClosed(status);
}

@Test
public void closeListener_deferredDeframerClose_stopDeliveryFalse_delaysCloseListener() {
ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
StatsTraceContext customStatsTraceCtx = new StatsTraceContext(new StreamTracer[] {mockTracer});
BaseTransportState state = new BaseTransportState(customStatsTraceCtx, transportTracer);
AbstractClientStream stream = new BaseAbstractClientStream(
allocator, state, new BaseSink() {}, customStatsTraceCtx, transportTracer);
stream.start(mockListener);

// Send partial message into deframer
byte[] data = new byte[] {0, 0, 0, 0, 2, 1}; // 2-byte frame, only 1 byte delivered
state.deframe(ReadableBuffers.wrap(data));

Status statusFalse = Status.CANCELLED.withDescription("deferred stopDelivery false");
state.transportReportStatus(statusFalse, false, new Metadata());

// Listener is not closed yet because deframer is mid-frame and waiting for complete frame
verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer, never()).streamClosed(any(Status.class));

// Request message and provide remaining byte of frame to complete deframer processing
stream.request(1);
state.deframe(ReadableBuffers.wrap(new byte[] {2}));
verify(mockTracer, never()).cancelled(any(Status.class));
verify(mockTracer).streamClosed(any(Status.class));
}

@Test
public void startFailsOnNullListener() {
AbstractClientStream stream =
Expand Down
49 changes: 49 additions & 0 deletions core/src/test/java/io/grpc/internal/StatsTraceContextTest.java
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
/*
* 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.internal;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;

import io.grpc.ClientStreamTracer;
import io.grpc.ServerStreamTracer;
import io.grpc.Status;
import io.grpc.StreamTracer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;

/** Unit tests for {@link StatsTraceContext}. */
@RunWith(JUnit4.class)
public class StatsTraceContextTest {

@Test
public void clientCancelled_notifiesClientStreamTracers() {
ClientStreamTracer clientTracer = mock(ClientStreamTracer.class);
ServerStreamTracer serverTracer = mock(ServerStreamTracer.class);

StatsTraceContext statsTraceCtx = new StatsTraceContext(
new StreamTracer[] {clientTracer, serverTracer});

Status cancelledStatus = Status.CANCELLED.withDescription("Client cancelled");
statsTraceCtx.clientCancelled(cancelledStatus);

verify(clientTracer).cancelled(cancelledStatus);
verifyNoInteractions(serverTracer);
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -846,6 +846,7 @@ public void cancel(Status reason) {
if (!internalCancel(serverStatus, serverStatus)) {
return;
}
statsTraceCtx.clientCancelled(reason);
serverStream.clientCancelled(reason);
streamClosed();
}
Expand Down
Loading
Loading