Add ability to replay last N items of a Multi on subscription, or emit something predefined if nothing has been emitted yet

[Ladicek]

Rx has a concept of BehaviorSubject and ReplayingSubject, which are quite similar and serve a very useful purpose: pre-seed a subscriber with last known value[s] / initial value[s] at the moment of subscription. I think we could unify them under an API like this:

Multi.createBy()
    .replayingAtMost(int)  // buffer size -- to imitate `BehaviorSubject`, use 1
    .withInitialValues(T... or Iterable<T> or perhaps even Multi<T>) // initial values to emit when nothing was emitted yet
    .stream(Multi<T>); // the stream we're wrapping

Calling it replayingAtMost to highlight that if you set 5 items to replay but only 3 items have been emitted, then only 3 will be replayed to a new subscriber. If you think that's intuitive enough, can shorten to just replaying.

WDYT?

[jponge]

Do you have any specific use-case in mind?

[Ladicek]

I filed this issue as a result of a Gitter discussion starting here: https://gitter.im/smallrye-io/community?at=608a65e5bdc0bc32e6258397

[jponge]

I went back thinking about this one.

BehaviorSubject / ReplayingSubject in RxJava are hot streams, so it does make sense to replay items there.

Now Mutiny Multi are cold streams in most cases, so that feature would require turning the wrapped Multi into a hot stream (using . toHotStream()). Depending on the use case one might want to do that... or not 😃 because the first subscriber triggers the source, and then all subscribers will simply join the stream broadcast.

Is it what you had in mind?

[Ladicek]

Yes. I don't think this feature would make much sense for cold streams, the entire feature implicitly assumes that there's a single broadcast that all subscribers listen to together.

[jponge]

(notes for myself and the interested reader)

• I've explored a broken draft where I'd collect items to replay.
• The first subscriber would make the wrapped multi broadcasting.
• Each subscriber would be plugged to the concatenation of (i) a stream of the last n items to replay, followed by the broadcasting stream.
• Aside from questions marks that I have over the behaviour of broadcast (need another issue for that), that approach is subject to items loss between the 2 streams concatenation.
• We likely need a full replay operator + supporting API here to have fine control over (i) observing events from the wrapped multi, and (ii) ensure that each subscriber gets a proper replay + stream. I would have preferred composing existing operators, but the semantics of replaying while another stream goes on aren't that trivial. But I may be missing something.

[jponge]

So I have something in draft in a scratchpad branch. How does that sound?

• Multi.createBy().replaying().upTo(3).ofMulti(upstream)
• Multi.createBy().replaying().ofMulti(upstream)
• Multi.createBy().replaying().ofSeedAndMulti(seed, upstream) (where seed is an Iterable<T>)
• Multi.createBy().replaying().upTo(10).ofSeedAndMulti(seed, upstream) (where seed is an Iterable<T>)

[jponge]

In that branch:

• upstream becomes a hot-stream (aka subscribed with Long.MAX_VALUE), and
• subscribers get to replay at their own pace (we honour each subscriber demand at all times)

[Ladicek]

Do I understand it correctly that the variants without upTo() will [maybe obtain the seed values and then] just join the broadcast? IOW, they are equivalent to upTo(0)?

When I created this issue, I intentionally suggested createBy().replayingAtMost(int) (or, in your parlance, createBy().replayingUpTo(int)) so that you can't skip the number of replayed items. If you didn't want to replay anything, just get the initial values and then join the broadcast, you'd have to say 0 explicitly, which perhaps looks weird, but then using a "replay" feature without actually replaying anything should look weird IMHO :-)

[jponge]

If you omit calling upTo(n) (n > 0) then you get to replay all events once you subscribe. If you do call upTo(n) then you get to replay the last n elements when you subscribe, and if the stream hasn't completed yet then you also receive the rest of the stream (like in a broadcast). The seed will be prepended to the stream, so wether you receive these events or not depends on how much you are replaying (bounded / unbounded) and at which point in time you subscribe.

…

On Thu, Mar 3, 2022, at 08:52, Ladislav Thon wrote: Do I understand it correctly that the variants without `upTo()` will [maybe obtain the seed values and then] just join the broadcast? IOW, they are equivalent to `upTo(0)`? When I created this issue, I intentionally suggested `createBy().replayingAtMost(int)` (or, in your parlance, `createBy().replayingUpTo(int)`) so that you can't skip the number of replayed items. If you didn't want to replay anything, just get the initial values and then join the broadcast, you'd have to say `0` explicitly, which perhaps looks weird, but then using a "replay" feature without actually replaying anything *should* look weird IMHO :-) — Reply to this email directly, view it on GitHub <#536 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAAGK2KSUU2W7ZO7RCVTKGLU6BVSXANCNFSM43ZXAYOA>. You are receiving this because you were assigned.Message ID: ***@***.***>

[Ladicek]

Ah so it's like upTo(Long.MAX_VALUE), roughly :-) OK, that makes sense.

[jponge]

Exactly! Long.MAX_VALUE is the default (aka replay everything).

…

On Thu, Mar 3, 2022, at 09:05, Ladislav Thon wrote: Ah so it's like `upTo(Long.MAX_VALUE)`, roughly :-) OK, that makes sense. — Reply to this email directly, view it on GitHub <#536 (comment)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAAGK2NTHDNJ3LLKSWPW7ODU6BXEBANCNFSM43ZXAYOA>. You are receiving this because you were assigned.Message ID: ***@***.***>