SplittingWriteEventsListener expects that an async source limits recursion depth to 1

[idelpivnitskiy]

SplittingWriteEventsListener expects the following sequence of HTTP messages:

meta-data → Start  → Payload → InProgress → trailers → End

If the source does not limit recursion to a depth of 1 when it delivers data (see PublisherFrom2TckTest and PublisherFrom3TckTest), the sequence of events can be any, depending on how WriteDemandEstimator requests items.

Steps to reproduce:

1. Change WriteDemandEstimators#newDefaultEstimator() to always return 1L from estimateRequestN.
2. Run ContentLengthAndTrailersTest

By playing with WriteDemandEstimator you can achieve any sequence of events. Examples:

meta-data → trailers → End → Start
meta-data  → Payload → trailers → End  → InProgress → Start
meta-data  → Payload → InProgress → Start → Payload → trailers → End  → InProgress

Because the sequent can be any, SplittingWriteEventsListener and CountingFlushStrategyProvider can not correctly detect message boundaries. As the result, data may hang in the socket buffer and will never be flushed.

This is less problematic when task-based offloading is enabled because each onNext will be on a different task thread, but causes issues when offloading is disabled.

Workaround: flushOnEach() strategy, which will trigger the flush for any item, regardless of the boundary. Users can opt-out from our internal flush strategy optimizations by using streaming API or applying transform operation on the HTTP message in a filter (it will reset the "aggregated" and "empty" flags).

Ways to reduce the likelihood of this issue for flushOnEnd():

• Limit recursion depth to 1 for all async sources and operators.

---

After this is fixed, revert the following patches:

• FromNPublisher: limit recursion depth to 1 #1653
• SingleConcatWithPublisher: limit recursion depth to 1 #1654