OnBackpressureXXX: support for common drain manager & fix for former concurrency bugs

[akarnokd]

This is quite a complex operator with lots of cases.

Properties:

1. If there aren't any elements queued up and nothing is requested but terminal event received, emit terminal event and quit.
2. If there are elements in the queue and a terminal flag, and at least the same amount is requested, deliver the events and the terminal event.
3. If more was requested and more became available just after the loop and before the synchronized block, keep looping.
   3.a) If more was requested but nothing is available or nothing was requested and something is available: quit and let either the onNext or request do the subsequent drain.
   3.b) If elements and termination was produced but not requested: quit and let the request do the drain
   3.c) If termination was requested and no elements produced: loop , emit terminal event and quit.

In table form:

Available | Terminated | Requested | Action | Reason
   yes         yes          yes       loop    can deliver available
   yes         yes          false     quit    can't deliver available
   yes         no           yes       loop    can deliver available
   yes         no           no        quit    can't deliver available
   no          yes          yes       loop    loop will deliver termination only and quit
   no          yes          no        loop    loop will deliver termination only and quit
   no          no           yes       quit    nothing to deliver
   no          no           no        quit    nothing to deliver

[akarnokd]

This is complicated stuff, maybe an (internal) helper class should be created to help with future producer-backpressure-consumer management.

[akarnokd]

I hope the new BackpressureDrainManager provides a common abstraction that let's build backpressure-providing operators that themselves do buffering of some sorts.

[benjchristensen]

Wow, this is some non-trivial stuff. Thanks for tackling this. It's going to take me a bit to grok it.

[benjchristensen]

Anyone else have the time and interest to also do a code review on the concurrency code in this? I'd appreciate more eyes than my own.

[benjchristensen]

In this use case why would we ever be unsubscribing early? We can emit an onError, but I don't see the need to decouple the subscription.

[akarnokd]

assertCapacity calls unsubscribe and would disrupt downstream.

[benjchristensen]

I think this code is good to merge. I've asked a variety of clarifying questions that I want to review with you before I merge and release.

[akarnokd]

Wrong copyright, should be Netflix

[akarnokd]

Re-trigger travis.

[zsxwing]

Need to add if (capacity != null) capacity.incrementAndGet(); here.

[zsxwing]

Could you change CRLFs to LFs in BackpressureDrainManager?

[zsxwing]

isTerminated, terminate, terminate(Throwable) are not used. Could you explain why adding them?

[zsxwing]

This is really a complex class. So I prefer to keep it as simple as possible.