Streams with flexible requestN semantic

[whyoleg]

• Reworked streams API using coroutine context element RequestStrategy for better control over requestN frames
• fixes Limit rate operator for streams #109

[whyoleg]

@yschimke
What do you think about it?
I've found, that it's really not so easy to integrate better backpressure control using just Flow and current RSocket interface. So I created new ReactiveFlow inherited from Flow (naming is TBD). And because only incoming Flow should be controlled by requestN (so on requester side it's requestStream and requestChannel output and on responder side requestChannel input), and for other cases it's not needed, and will just add complexite, I've decided to split RSocket interface for requester and responder.
So, do you think, that this approach is OK, and what do you think about RequestStrategy API?

and yes, it's still WIP, need more tests and fixes may be, but overall works

[yschimke]

@whyoleg I think I'm the wrong reviewer for this. You should try to get some of the Pivotal folk with more experience across rsocket-java and spring reactor. There have been a lot of discussions in the past on reactors infinite and cancel defaults and limits, and I have simple requirements to be honest now.

In FB days it was more relevant to me since there was a polling mechanism in the background and a real cost to every additional emmitted result, but infinite and cancel is mostly fine for what I'm doing now e.g. rsocket-demo with a stream of tweets (line separated http response) or with my own backends being purely event based anyway.

[yschimke]

Outside this review...

This sort of pattern, reuse of a payload probably deserves some nicer more natural mechanism. This is likely a source of bugs under error conditions which will be missed on the happy path. How do we improve this? e.g. payload.permanent()?

[whyoleg]

and how payload.permanent() will work? have just possibility to reuse data? and release after all?
@altavir has much experience working with kotlin with IO, and now researching adding rsocket-kotlin to his own projects. Maybe hi have some ideas on better payload handling?

[OlegDokuka]

Just from my side. We do allocate a ByteBuf once, and then retain it every time we use it in benchmarks. Sounds like a copy but not really, thus not sure Kotlin IO can do the same

[yschimke]

Or maybe @elizarov and work out what the extension methods to ReactiveFlow should be. I think ideally none new inside rsocket-kotlin. We would just react to the usage and contract of ReactiveFlow (edit: I thought this was part of kotlin, later answer corrects this). And use concepts the end users are already familiar with externally.

IIRC That was essentially the decision with rsocket-java and spring reactor. Not to build our own terminology, but to keep using the operators from the underlying framework.

[whyoleg]

The idea behind ReactiveFlow is to just store some RequestStrategy which will allow for better control of requestN frames and in current implementation, it has only one realization PrefetchStrategy(requestSize, requestOn) - which will request requestSize elements, when requestOn elements are left for collecting in Flow. So, the simplest one.
Im not sure, that something else should be added for now. From user perspective, it's even may be not needed, and default strategy (requestSize=64, requestOn=16) will be enough.

We need to add such logic, with one new type (ReactiveFlow), as Flow only works with collecting one by one, and even take(5) will work as collect by one until 5 elements collected.

[yschimke]

I'm not a huge fan of defining our own ReactiveFlow

[elizarov]

If all you need is to store a request strategy, then I can suggest looking into using a context for this kind of storage. This way you can have a sensible default that will add zero overhead (and zero new concepts) and rely on the existing context-passing support in both Reactor and Kotlin Flow to change this request strategy if needed.

[whyoleg]

@elizarov cool idea, I will try to do that! thx!

[OlegDokuka]

as discussed, lets split these changes into two PRs:

1. For RequestN via flowOn
2. RSocket interface API changes

[OlegDokuka]

LGTM with some minor comments

[OlegDokuka]

Just wondering why do we need atomic here?

[whyoleg]

it needed to allow mutation on K/N if we freeze. need to check, may be it's really not needed, here.
On K/JVM and K/JS it's not needed, as functions inside RequestStrategy.Element will never be called concurrently

[OlegDokuka]

yes, please double-check. In general, it looks like all the requests will be done within the same context, hence the context should do all the stuff to keep seriality, hence no atomic is needed

[whyoleg]

atomic removed, not needed, thx for pointing!

[OlegDokuka]

but here, this atomic is used in non-atomic way, Am I missing something?

[OlegDokuka]

LGTM 🔥