private void subscribe(SubscriptionImpl subscription, Subscriber<Object> subscriber) {
        try {
            subscribe0(subscription.executor(), subscription.options());
            // 'invokedOnSubscribe' should be set after 'subscribe0()' is completed.
            // 'onComplete()' could be invoked by a subclass which overrides 'subscribe0()' to subscribe
            // to other Publishers.
            maybeInvokeOnSubscribe(subscription, subscriber);
            if (!queue.isEmpty()) {
                notifySubscriber0();
            }
        } catch (Throwable t) {
            if (setState(State.OPEN, State.CLEANUP) || setState(State.CLOSED, State.CLEANUP)) {
                notifySubscriberOfCloseEvent(subscription, newCloseEvent(t));
                throwIfFatal(t);
            } else {
                throwIfFatal(t);
                logger.warn("Subscriber.onSubscribe() should not raise an exception. subscriber: {}",
                            subscriber, t);
            }
        }
    }
Creates a new instance.

@deprecated Use {@link StreamMessage#streaming()} instead.
core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessage.java