private void cleanupObjects(@Nullable Throwable cause) {
        for (;;) {
            final Object e = queue.poll();
            if (e == null) {
                break;
            }

            if (e instanceof CloseEvent) {
                continue;
            }

            if (e instanceof AwaitDemandFuture) {
                if (cause == null) {
                    cause = ClosedStreamException.get();
                }
                ((CompletableFuture<?>) e).completeExceptionally(cause);
                continue;
            }

            try {
                @SuppressWarnings("unchecked")
                final T obj = (T) e;
                onRemoval(obj);
            } finally {
                StreamMessageUtil.closeOrAbort(e, cleanupCause);
            }
        }
    }
Tries to close the stream with the specified {@code cause}.

@return {@code true} if the stream has been closed by this method call.
        {@code false} if the stream has been closed already by another party.
core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessage.java