private void doUnsubscribe(@Nullable Throwable cause) {
            if (!processor.downstreamSubscriptions.remove(this)) {
                return;
            }

            final Subscriber<? super T> subscriber = this.subscriber;
            clearSubscriber();

            final CompletableFuture<Void> completionFuture = whenComplete();
            if (cause == null) {
                try {
                    subscriber.onComplete();
                    completionFuture.complete(null);
                } catch (Throwable t) {
                    completionFuture.completeExceptionally(t);
                    throwIfFatal(t);
                    logger.warn("Subscriber.onComplete() should not raise an exception. subscriber: {}",
                                subscriber, t);
                } finally {
                    processor.cleanupIfLastSubscription();
                }
                return;
            }

            try {
                if (notifyCancellation || !(cause instanceof CancelledSubscriptionException)) {
                    subscriber.onError(cause);
                }
                completionFuture.completeExceptionally(cause);
            } catch (Throwable t) {
                final Exception composite = new CompositeException(t, cause);
                completionFuture.completeExceptionally(composite);
                throwIfFatal(t);
                logger.warn("Subscriber.onError() should not raise an exception. subscriber: {}",
                            subscriber, composite);
            } finally {
                processor.cleanupIfLastSubscription();
            }
        }
{@link CancelledSubscriptionException} if cancelled. {@link Throwable} if aborted.
core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessageDuplicator.java