Operator requestBatching to change the in-flight event amounts

[akarnokd]

I'm proposing an operator that can change (lower) the in-flight request amount for both an unlimited downstream and for a backpressuring downstream.

I'm not certain about its use cases besides forcing a slow-path emission in a stream where the Subscriber requests Long.MAX_VALUE, for example with the rewritten merge of #2928 where one merges long streams asynchronously, the internal serialization buffer can get quite large. The use of this operator would switch it back to a slower-path merging where bounded buffers are employed.

[benjchristensen]

I'm not certain about its use cases

I'm struggling to understand use cases for this since this would naturally occur from async operators.

I haven't thought through it enough, but my immediate concern is that we'd have edge cases where we hang the system as a consumer thinks it has requested values, but the producer is not emitting as the "batcher" is waiting until some threshold to send them up.

besides forcing a slow-path emission in a stream where the Subscriber requests Long.MAX_VALUE, for example with the rewritten merge

The merge case does not seem to be a use case for this, as even with this it would still cause unbounded buffer growth because merge in #2928 will swallow all data thrown at it, regardless of how requests are batched. It would just keep requesting up ... and if it didn't, then it has correctly implemented backpressure and doesn't need this.

Additionally, we would have to apply this operator to every stream being merged, which should not be needed. Instead merge just needs to correctly do back pressure, like the current implementation does.

[benjchristensen]

@akarnokd do you still want to pursue this?

[akarnokd]

No.