@Test
    void withPooledObjects() {
        final HttpData httpData = httpData(0);
        final ByteBuf byteBuf = httpData.byteBuf();
        final StreamWriter<HttpData> publisher = StreamMessage.streaming();
        final StreamMessageDuplicator<HttpData> duplicator =
                publisher.toDuplicator(ImmediateEventExecutor.INSTANCE);
        publisher.write(httpData);
        publisher.close();

        final AtomicBoolean completed = new AtomicBoolean();
        duplicator.duplicate().subscribe(new Subscriber<HttpData>() {
            @Nullable
            Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                // Cancel the subscription when the demand is 0.
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(HttpData d) {
                assertThat(byteBuf.refCnt()).isEqualTo(2);
                subscription.cancel();
                d.close();
                completed.set(true);
            }

            @Override
            public void onError(Throwable throwable) {
                // This is not called because we didn't specify NOTIFY_CANCELLATION when subscribe.
                fail("unexpected onError()", throwable);
            }

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

        await().untilAsserted(() -> assertThat(completed).isTrue());
        assertThat(byteBuf.refCnt()).isOne();
        duplicator.close();
        assertThat(byteBuf.refCnt()).isZero();
    }
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