@Test
    void cancellationIsNotPropagatedByDefault() {
        final StreamWriter<Integer> delegate = StreamMessage.streaming();
        final PublisherBasedStreamMessage<Integer> p = new PublisherBasedStreamMessage<>(delegate);

        p.subscribe(new Subscriber<Integer>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.cancel();
            }

            @Override
            public void onNext(Integer integer) {
                fail("onNext() invoked unexpectedly");
            }

            @Override
            public void onError(Throwable t) {
                fail("onError() invoked unexpectedly");
            }

            @Override
            public void onComplete() {
                fail("onComplete() invoked unexpectedly");
            }
        });

        // We do call onError(t) first before completing the future.
        await().untilAsserted(() -> assertThat(p.whenComplete()).isCompletedExceptionally());
    }
Tests if {@link PublisherBasedStreamMessage#abort()} prohibits further subscription.
core/src/test/java/com/linecorp/armeria/common/stream/PublisherBasedStreamMessageTest.java