WriteStreamSubscriber: respect termination of the publisher

[idelpivnitskiy]

Motivation:

1. Protocol can signal when the effective write of the message is complete and no more items are expected. In this case, WriteStreamSubscriber will do request(MAX_VALUE) to drain the remaining (likely empty) items from the publisher and terminate subscriber. However, premature termination of the completable subscriber breaks Reactive Streams control flow. Users of
   connection.write(publisher) can compose more operators on the returned Completable and they expect that the Completable terminates only after the passed publisher terminates too. The publisher can have more business logic after it emits all items. For example, service can read request payload body after it returns the full response.
2. If it waits for continuation but protocol signals it's not interested in more data, publisher won't know about it.

Modifications:

• On channelOutboundClosed(), do not attempt to terminate subscriber. Instead, mark the state as SOURCE_OUTBOUND_CLOSED and wait for termination of the publisher;
• If channelClosed(...) was invoked after channelOutboundClosed(), do not fail the subscriber. Instead, wait for termination of the publisher;
• When terminateSource() is invoked, cancel the current subscription to let the publisher know we are not interested in more items. For HTTP client it won't have any actual effect for request payload body because we use Single.concat(Publisher) with deferSubscribe option, but it's necessary to propagate signal for other use-cases;
• Adjust tests for new behavior;
• Increase test coverage for WriteStreamSubscriber;

Result:

Terminal signals are correctly propagated through the RS chain when users use connection.write(publisher).

[idelpivnitskiy]

Required for #2367.

[Scottmitch]

is this assertion necessary? wondering if we need to prevent any write-after-close/terminated scenarios as we may want to pass on the object anyways (reference counted, ordered promos completion, ..)

[idelpivnitskiy]

Not really, invocation of this method is currently guarded by isWritable() check:

    void doWrite(Object msg) {
        // Ignore onNext if the channel is already closed.
        if (promise.isWritable()) {
            long capacityBefore = channel.bytesBeforeUnwritable();
            promise.writeNext(msg);

Added this assection just in case the contract changes, to make sure we handle writes correctly as it will likely will also required changes in AllWritesPromise.