Reinstantiate backpressure propagation to NettyInbound

[mp911de]

With a recent change, we no longer propagate the backpressure from the R2DBC SPI (e.g. Result.map(…)) to NettyInbound. Previously, we used windowing and passed the Flux<Message> directly to the consumer. Now we're using handle(…) and send the messages to the conversation receiver instead of to a window that is directly connected to NettyInbound.

The missing backpressure propagation can lead to overflows.

[Squiry]

Well it's not like backend will stop message producing if we propagate it to NettyInbound, so why should we care?

[mp911de]

Not reading messages from the channel leaves the messages in the receive buffers. If these buffers are full, then the sending side does not receive acknowledgments and is required to slow down. This effect is called TCP Backpressure and part of TCP Flowcontrol. See also Wikipedia. The subscriber is guaranteed to not overload and that is a protection for the client application.

In blocking I/O we get the same level of protection if we properly set up e.g. the JDBC ResultSet fetch size. As long as ResultSet.next() is not called, results are not read from I/O.

[Squiry]

Well reactor netty kind of doesn't care too: it will store 'em all in a queue, not in a tcp buffer.

[mp911de]

Right, Reactor Netty can buffer a few messages. If FluxReceive sees there is no demand, it disables auto-read from the channel and that is how additional TCP packets are kept outside of the JVM process.

[mp911de]

@Squiry I took your commits and refactored BackendMessageSubscriber and Conversation to make it nicer. As of now, all tests run green.

I'm still worried by the impact this change may have and all the side-effects that can come with it.