Surfacing ReplaySubject timestamps in materialized events

[joshribakoff]

Feature Request

replay subject (when using windowTime) records timestamps each time you next a value onto it: https://github.com/ReactiveX/rxjs/blob/master/src/internal/ReplaySubject.ts#L79 I'd like to be able to surface these timestamps upon subscription

Is your feature request related to a problem? Please describe.

I am building a library for using RxJS with React. I did some exploring on an issue rx-store/rx-store#12 which is basically to be able to subscribe to a subject in a parent React component, and "next" onto that subject from a child component. Subscriptions in RxJS are not idempotent and therefore must happen in commit phase React lifecycle methods. Furthermore, React mounts & fires componentDidMount for children components before parent components, so if children "next" onto a subject before the parent component subscribes to that subject, it will not receive the values. Because of this, a ReplaySubject would be the perfect escape hatch.

To frame it outside the context of React, you may want to start some task & emit metrics / progress onto some subject... but you may not be able to subscribe a side effect to that subject until after it has started emitting, for various different reasons.

Describe the solution you'd like

I'd make a PR that:

• renames nextInfiniteTimeWindow to nextWithoutTime
• renames nextTimeWindow to nextWithTime
• add a 4th, optional arg to the constructor called materialize (bool)
• ]changes monkey patching of next function ](

  rxjs/src/internal/ReplaySubject.ts

  Line 60 in cd7d649

  this.next = this.nextInfiniteTimeWindow;

  ) to check if windowTime is used, or if materialize is used instead.
• passing the materialize boolean as true will cause the ReplaySubject to next the Replay Event (object w/ value & timestamp) rather than just the value.

Describe alternatives you've considered

• Mapping values to timestamps will not work, as the issue is specifically related to time stamping things prior to subscription
• Schedulers again cannot solve this, either the burden is on the end user to supply the timestamps, or the replay subject must do it
• materialize operator is not the right solution either, this is specific to ReplaySubject, which doesn't deal with timestamps for errors/completions currently either. Notification type should not be modified.
• Forcing the end user to put their own timestamps into the value is a workaround, but not ideal, because ReplaySubject already calls now() and it'd be wasteful & not clean to have the end user call it a second time.

[kwonoj]

I'm slightly inclined to say that timestamp is an internal implementation detail. I can see use case of your feature proposal, while would like to understand how it can be used in general, esp emitting different materialized kind-of Notification object by config doesn't seem to usual practices in our API surfaces. For me this sounds like more you need to create your own type of subject instead of extending existing replaysubjet behavior - mind fill in bit more details which cases consumer would need materialized types of emission instead of original value other than your specific usecase?

[joshribakoff]

I'm slightly inclined to say that timestamp is an internal implementation detail.

It's off by default in my PR :) Most of the time, you're right, it is. In my case it's not an implementation detail. I don't want my code to call performance.now() twice for each event, if RxJs is already doing it internally. Especially if I'm processing a noisy stream, doubling the number of calls to the system clock is unnecessary bloat.

mind fill in bit more details which cases consumer would need materialized types of emission instead of original value other than your specific usecase?

What's the use case for replay subject in the first place, exactly?

As an end user I expected the materialize() operator to do this already, and ended up digging through the source before I understood why it does not (now I understand why it does not).

I'd argue that if there is a use case for ReplaySubject at all, to replay values in general, it seems like there would be a use case for knowing when the values were originally received.

In React, the reason it works this way is because React can start rendering a tree, and abort (concurrent mode) to work on a higher priority update. Because children mount first, its impossible to create a subject in the parent, safely subscribe to it, and pass it down to a child to "next" onto when it mounts. Subscription must happen after child mounts, which is when the parent mounts. If subscription were to happen before the parent mounts, and react interrupts & resumes rendering, subscription would happen 2x, and RxJS subscriptions are not idempotent.

To contrive other use cases, I guess, you could for example, use the delay() operator in creative ways to cause the original values to be emitted over time with the same delays in which the original ReplaySubject received them. There could be some implications for "time travel" there, or other contrived use cases I'd have to think more deeply about. I can't think of any use case for replay subject at all, other than cases where it might also be useful to capture the timestamp.

For me this sounds like more you need to create your own type of subject instead of extending existing replaysubjet behavior

This is certainly possible, but there's already 4 types of subjects, and imposing a 5th type on users of my library is not ideal, the whole idea of my library is to just use RxJS everywhere & ideally not make a wrapper where I can avoid it, my library is really more like a pattern :) Wrapping RxJS is a last resort. In fact, without looking in too much detail, it almost feels like BehaviorSubject & ReplaySubject naively do the same thing & should also be consolidated? I'd potentially need to make a 6th, "timestamp aware" behavior subject, and force that complexity on my end users too, as that subject also replays [a] value(s).

Is the trade off here "constructor overloads" (a subject who's constructor is too complex & variadic) vs "separate constructors" (having too many subjects, which is hard to learn/remember)? @staltz identified this tradeoff a while ago in #73 and suggested Behavior could be implemented with Replay. Having both constructors make sense but ideally they re-use logic under the hood.

Is there some other out of the box solution to emit onto subjects before subscription & not loose the timing, and not introduce a 5th or 6th type of subject?

[kwonoj]

Few things at least:

• I'd argue that if there is a use case for ReplaySubject at all, to replay values in general, it seems like there would be a use case for knowing when the values were originally received.

I'd like to point out those time stamping is only to support specific variant of replay cases for time based window. For all other replaysubject variant, it is not required to record timestamp, as well as internal behavior of timestamping can change any time since replaysubject does not have any contract around internal timestamp for old emissions. I'd argue use case for knowing when the values were originally received. for all kinds of replaysubject should be treated specific custom case worth userland operator.

• number of subject in core

yes, there are some numbers and there were historical reason (parity to original .net binding implementation) and core team actively discussing possibilities to reduce numbers.

• Is the trade off here "constructor overloads"

No, what we care more is about general amount of api surface itself - whether it's exposed via constructor or different manner of subject implementation, it doesn't change we are increasing new behavior surface in core and we want to strictly limit (and reduce) it.

• Is there some other out of the box solution to emit onto subjects before subscription & not loose the timing, and not introduce a 5th or 6th type of subject?

I doubt we have one.

[joshribakoff]

Yeah, so I'm all for minimizing the API, and I'd like to help with that, but like you said this is a use case that isn't supported at all now, and IMO its quite minimal & fully backwards compatible.

For all other replaysubject variant, it is not required to record timestamp

I see a use case for any subject that does any kind of replaying, to be able to get the original "next" timestamp. I would actually add this to all of them. I could update the others, if we did decide to do this issue.

for all kinds of replaysubject should be treated specific custom case worth userland operator.

Can't be an operator, the operator will sample the time from it's scheduler when the value is replayed, not when the replay subject received it. Currently the replay subject does not surface the timestamp, that information is lost, it is gone by the time any operator's observer sees the value. this must be done in userland, or in the replay subject.

internal behavior of timestamping can change any time since replaysubject does not have any contract around internal timestamp for old emissions

I'm not really sure what is meant by this. The contract would be that it surfaces the value returned by the timestampProvider. I know it doesn't have a contract now, because it doesn't have this feature, but that would be the contract I'm proposing we'd add

[joshribakoff]

Maybe the right solution is to allow giving the replay subject a pipeline of operators somehow? Thoughts?

This maps in the timestamp at subscription time:

import { ReplaySubject } from 'rxjs'; 
import { map } from 'rxjs/operators';


const source$ = new ReplaySubject()

const time = Date.now()
console.log('nexting',{time})

source$.next(time)

setTimeout(()=>{
  source$
    .pipe(map(val=>({val, time: Date.now()})))
    .subscribe(x => console.log('subscriber saw: ',x, ' at ', Date.now()));
}, 5000)

Instead we could do it at "next"ing time:

import { ReplaySubject } from 'rxjs'; 
import { map } from 'rxjs/operators';

const pipe = [
   map(val=>({val, time: Date.now()}))
]
const source$ = new ReplaySubject( ... pipe ...) // pipe is injected somewhere

const time = Date.now()
console.log('nexting',{time})

source$.next(time)

setTimeout(()=>{
  source$
    .subscribe(x => console.log('subscriber saw: ',x, ' at ', Date.now()));
}, 5000)

This would potentially allow removing the ad hoc buffer / time logic, and open up a plethora of new use cases for replay subjects. There'd be no longer more than 1 simple subject, and it's behavior is composable. Eg. choosing whether it replays, whether the replay is buffered or windowed, would all be done through Rx operators instead of ad hoc args.

[joshribakoff]

Actually I think shareReplay operator is what I wanted, it subscribes to my subject, creates an internal ReplaySubject, and if I understand right it should allow me to map in a timestamp with the map operator right before this shareReplay operator. If this is in fact possible with operators alone today, why do we need ReplaySubject at all? (Why isn’t it internal only?) i don’t think this is possible with operators today, as shareReplay appears to require an initial subscription for a second late subscriber, whereas my issue is with a late initial subscriber

Also in my use case I want to be able to replay the events once, ideally after I’m subscribed I don’t want it to keep all the old events in memory... I’m curious about the best way to solve this.

Ideally the shareReplay operator ReplaySubject would be able to use some other stream to trigger the buffering off, such as when the parent has mounted, and events no longer need to be buffered...

In any case it seems pretty important to me to be able to create subjects in react components and next onto them in the children component, and to do so without losing access to the timings information

[cartant]

Closing this because the related PR - #5516 - was closed.