Repository: apache/dubbo
File: dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractTripleClientStream.java (Java)

Code before:

    /**
     * Called by InitOnReadyQueueCommand to trigger the initial onReady notification.
     */
    public void triggerInitialOnReady() {
        notifyOnReady(true);
    }

    /**
     * notify listener when stream becomes ready
     *
     * @param forceNotify if true, always trigger onReady (for initial notification);
     *                    if false, only trigger when state changes from "not ready" to "ready"
     */
    private void notifyOnReady(boolean forceNotify) {
        boolean wasReady = lastReadyState;
        boolean isNowReady = isReady();
        lastReadyState = isNowReady;

        // Trigger onReady if:
        // 1. forceNotify is true (initial notification, spurious is OK), or
        // 2. state changes from "not ready" to "ready"
        if (forceNotify || (!wasReady && isNowReady)) {
            listener.onReady();
        }
    }

    class ClientTransportListener extends AbstractH2TransportListener implements H2TransportListener {

        private TriRpcStatus transportError;
        private DeCompressor decompressor;
        private boolean headerReceived;
        private Http2Headers trailers;

        void handleH2TransportError(TriRpcStatus status) {
            writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.NO_ERROR));
            rst = true;
            finishProcess(status, null, false);
        }

        void finishProcess(TriRpcStatus status, Http2Headers trailers, boolean isReturnTriException) {
            final Map<CharSequence, String> reserved = filterReservedHeaders(trailers);
            final Map<String, Object> attachments =
                    headersToMap(trailers, () -> reserved.get(TripleHeaderEnum.TRI_HEADER_CONVERT.getKey()));
            final TriRpcStatus detailStatus;
            final TriRpcStatus statusFromTrailers = getStatusFromTrailers(reserved);
            if (statusFromTrailers != null) {
                detailStatus = statusFromTrailers;
            } else {
                detailStatus = status;
            }

Review comment (bug):
This change introduces backpressure race condition handling logic, but there are no new tests to verify the race condition is actually fixed. Tests should be added to verify that concurrent calls to notifyOnReady correctly track state transitions and don't result in duplicate or lost notifications, especially when onWritabilityChanged and sendMessage callbacks race.

Code after:
    /**
     * Called by InitOnReadyQueueCommand to trigger the initial onReady notification.
     */
    public void triggerInitialOnReady() {
        notifyOnReady(true);
    }

    /**
     * notify listener when stream becomes ready
     *
     * @param forceNotify if true, always trigger onReady (for initial notification);
     *                    if false, only trigger when state changes from "not ready" to "ready"
     */
    private synchronized void notifyOnReady(boolean forceNotify) {
        boolean wasReady = lastReadyState;
        boolean isNowReady = isReady();
        lastReadyState = isNowReady;

        // Trigger onReady if:
        // 1. forceNotify is true (initial notification, spurious is OK), or
        // 2. state changes from "not ready" to "ready"
        if (forceNotify || (!wasReady && isNowReady)) {
            listener.onReady();
        }
    }

    class ClientTransportListener extends AbstractH2TransportListener implements H2TransportListener {

        private TriRpcStatus transportError;
        private DeCompressor decompressor;
        private boolean headerReceived;
        private Http2Headers trailers;

        void handleH2TransportError(TriRpcStatus status) {
            writeQueue.enqueue(CancelQueueCommand.createCommand(streamChannelFuture, Http2Error.NO_ERROR));
            rst = true;
            finishProcess(status, null, false);
        }

        void finishProcess(TriRpcStatus status, Http2Headers trailers, boolean isReturnTriException) {
            final Map<CharSequence, String> reserved = filterReservedHeaders(trailers);
            final Map<String, Object> attachments =
                    headersToMap(trailers, () -> reserved.get(TripleHeaderEnum.TRI_HEADER_CONVERT.getKey()));
            final TriRpcStatus detailStatus;
            final TriRpcStatus statusFromTrailers = getStatusFromTrailers(reserved);
            if (statusFromTrailers != null) {
                detailStatus = statusFromTrailers;
            } else {
                detailStatus = status;
            }
            listener.onComplete(detailStatus, attachments, reserved, isReturnTriException);