@Override
    public void subscribe(Subscriber<? super T> subscriber, EventExecutor executor,
                          SubscriptionOption... options) {
        requireNonNull(subscriber, "subscriber");
        requireNonNull(executor, "executor");
        requireNonNull(options, "options");

        final InnerSubscriber<T> innerSubscriber = new InnerSubscriber<>(subscriber, options, executor, this);
        final OuterSubscriber<T> outerSubscriber = new OuterSubscriber<>(innerSubscriber, executor);
        if (outerSubscriberUpdater.compareAndSet(this, null, outerSubscriber)) {
            subscriber.onSubscribe(innerSubscriber);
            sources.subscribe(outerSubscriber, executor, options);
        } else {
            subscriber.onSubscribe(NoopSubscription.get());
            subscriber.onError(new IllegalStateException("subscribed by other subscriber already"));
        }
    }
Relay items in order from a {@link StreamMessage} of {@link StreamMessage}s as a single {@link StreamMessage}
source.
core/src/main/java/com/linecorp/armeria/common/stream/ConcatPublisherStreamMessage.java