private boolean doSignalSingle(SignalQueue signals) {
            if (inOnNext) {
                // Do not let Subscriber.onNext() reenter, because it can lead to weird-looking event ordering
                // for a Subscriber implemented like the following:
                //
                //   public void onNext(Object e) {
                //       subscription.request(1);
                //       ... Handle 'e' ...
                //   }
                //
                // We do not need to worry about synchronizing the access to 'inOnNext' because the subscriber
                // methods must be on the same thread, or synchronized, according to Reactive Streams spec.
                return false;
            }

            if (cancelledOrAborted != null) {
                // Stream ended due to cancellation or abortion.
                unsubscribe(cancelledOrAborted);
                return false;
            }

            if (offset == processor.upstreamOffset) {
                // The subscriber read all signals published so far.
                return false;
            }

            final Object signal = signals.get(offset);

            if (signal instanceof CloseEvent) {
                // The stream has reached at its end.
                offset++;
                unsubscribe(((CloseEvent) signal).cause);
                return false;
            }

            for (;;) {
                final long demand = this.demand;
                if (demand == 0) {
                    return false;
                }

                if (demand != Long.MAX_VALUE && !demandUpdater.compareAndSet(this, demand, demand - 1)) {
                    // Failed to decrement the demand due to contention.
                    continue;
                }

                offset++;
                @SuppressWarnings("unchecked")
                T obj = (T) signal;
                try {
                    if (obj instanceof HttpData) {
                        final HttpData data = (HttpData) obj;
                        if (data.isPooled()) {
                            if (withPooledObjects) {
                                final ByteBuf byteBuf = data.byteBuf(ByteBufAccessMode.RETAINED_DUPLICATE);
                                @SuppressWarnings("unchecked")
                                final T retained = (T) HttpData.wrap(byteBuf)
                                                               .withEndOfStream(data.isEndOfStream());
                                obj = retained;
                            } else {
                                final ByteBuf byteBuf = data.byteBuf();
                                @SuppressWarnings("unchecked")
                                final T copied = (T) HttpData.copyOf(byteBuf)
                                                             .withEndOfStream(data.isEndOfStream());
                                obj = copied;
                            }
                        }
                    }
                } catch (Throwable thrown) {
                    // If an exception such as IllegalReferenceCountException is raised while operating
                    // on the ByteBuf, catch it and notify the subscriber with it. So the
                    // subscriber does not hang forever.
                    unsubscribe(thrown);
                    return false;
                }

                if (processor.isClosed() && processor.duplicator.unsubscribed == 0) {
                    if (++processor.downstreamSignaledCounter >= REQUEST_REMOVAL_THRESHOLD) {
                        // don't need to use AtomicBoolean cause it's used for rough counting
                        processor.downstreamSignaledCounter = 0;
                        int minOffset = Integer.MAX_VALUE;
                        for (DownstreamSubscription<?> s : processor.downstreamSubscriptions) {
                            minOffset = Math.min(minOffset, s.offset);
                        }
                        signals.requestRemovalAheadOf(minOffset);
                    }
                }

                inOnNext = true;
                try {
                    subscriber.onNext(obj);
                } catch (Throwable t) {
                    unsubscribe(t);
                    throwIfFatal(t);
                    logger.warn("Subscriber.onNext({}) should not raise an exception. subscriber: {}",
                                obj, subscriber, t);
                    return false;
                } finally {
                    inOnNext = false;
                }
                return true;
            }
        }
{@link CancelledSubscriptionException} if cancelled. {@link Throwable} if aborted.
core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessageDuplicator.java