@Override
        public void onNext(T o) {
            U filtered;
            try {
                filtered = filter(o);
            } catch (Throwable ex) {
                // onError(ex) should be called before upstream.cancel() to deliver the cause to downstream.
                // upstream.cancel() may close downstream with CancelledSubscriptionException before sending
                // the actual cause.
                onError(ex);

                assert upstream != null;
                upstream.cancel();
                return;
            }

            if (completed) {
                // onError(Throwable) or onComplete() has been called in filter().
                StreamMessageUtil.closeOrAbort(filtered);
                return;
            }
            if (!subscribedWithPooledObjects) {
                filtered = PooledObjects.copyAndClose(filtered);
            }
            delegate.onNext(filtered);
        }
A callback executed when this {@link StreamMessage} is canceled by the {@link Subscriber}.
core/src/main/java/com/linecorp/armeria/common/stream/FilteredStreamMessage.java