@Test
    void notifyCancellation() {
        final ByteBuf data = newPooledBuffer();
        final StreamWriter<HttpData> publisher = StreamMessage.streaming();
        final StreamMessageDuplicator<HttpData> duplicator =
                publisher.toDuplicator(ImmediateEventExecutor.INSTANCE);
        publisher.write(HttpData.wrap(data));
        publisher.close();

        final AtomicBoolean completed = new AtomicBoolean();
        duplicator.duplicate().subscribe(new Subscriber<HttpData>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.cancel();
            }

            @Override
            public void onNext(HttpData d) {
                fail("unexpected onNext(): %s", d);
            }

            @Override
            public void onError(Throwable t) {
                assertThat(t).isInstanceOf(CancelledSubscriptionException.class);
                completed.set(true);
            }

            @Override
            public void onComplete() {
                fail("unexpected onComplete()");
            }
        }, NOTIFY_CANCELLATION);

        await().untilAsserted(() -> assertThat(completed).isTrue());
        duplicator.close();
    }
A test for the {@link SignalQueue} in {@link DefaultStreamMessageDuplicator}.
Queue expansion behaves differently when odd/even number head wrap-around happens.
core/src/test/java/com/linecorp/armeria/common/stream/StreamMessageDuplicatorTest.java