void notifySubscriber(SubscriptionImpl subscription, CompletableFuture<?> completionFuture) {
            if (completionFuture.isDone()) {
                // Notified already
                return;
            }

            final Subscriber<Object> subscriber = subscription.subscriber();
            Throwable cause = this.cause;
            if (cause == null && subscription.cancelRequested()) {
                cause = CancelledSubscriptionException.get();
            }

            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);
                }
            } else {
                try {
                    if (subscription.shouldNotifyCancellation() ||
                        !(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);
                }
            }
        }
Replaces the subscriber with a placeholder so that it can be garbage-collected and
we conform to the Reactive Streams specification rule 3.13.
core/src/main/java/com/linecorp/armeria/common/stream/CancellableStreamMessage.java