Rate Limit Improvements

[rstoyanchev]

Rate limiting has been through a few iterations starting with the original LimitableRequestPublisher that actually imposed no upper limit and was exposed to issues like #514, later replaced by RateLimitableRequestPublisher in #672 which works like Flux#rateLimit but without batching up requests below the upper limit and therefore not requiring an internal queue. In #736 this is further refactored to RateLimitableRequestSubscriber. There is also a PR against Reactor Core #1879 creating a rateLimit variant based on the work in RSocket but with a more evolved implementation.

Rate limiting is important and probably should be more pluggable as well as visible in the RSocket API rather than being completely embedded. This could be an interceptor for example with a default but configurable upper limit but also possible to customize, replace, or completely turn off.

This would allow varying the implementation. For example using Flux#rateLimit to start could be a dead simple, default implementation, but it would also allow for other strategies too. Further options to consider now or eventually would be to vary rate limiting by specific streams.

Considering that most RSocket applications are likely using Reactor to compose logic, it's important to have rate limiting close to the transport because it may not be easy to apply rateLimit in application code, e.g. if subsequent code needs to apply async serialization of Objects to byte buffers which could bring in operators like flatMap with their own prefetch.