Discussing the future of schedulers

[benlesh]

Schedulers exist so we can exercise more control over when subscription occurs and when values are observed. They also provide a mechanism we can hook to get more deterministic tests that run very fast.

A few issues with the current Scheduler set up

1. It's hard to build a test suite for people.

• do we need a global default scheduler? Should we just pass one around?
• if an operator defaults to a particular scheduler, do we patch it to make it use a test scheduler during test scenarios?

2. It's a large and complicated part of the library
3. It's a little-understood part of the library
4. Scheduler arguments aren't symmetric with proposed Observable APIs, like Observable.of
5. Still doesn't make it much easier to test things like animationFrame or the like.
6. When a scheduler is needed, it must be passed around and provided to every Observable creation call and time-related operator.

Questions

• In practice, is it better to have an Observable.of(1, 2, 3, async) or Observable.asyncOf(1, 2, 3)? Observable.of(1, 2, 3, asap) or Observable.microtaskOf(1, 2, 3) (name totally bike-sheddable)
• Could we potentially just use setTimeout or Promise then, and patch them temporarily at test time? It seems like that might make it easier to build test suites. It might also reduce the size of the library.
• How much do people really find the need to use Schedulers? How valuable are they if we found another way to deterministically test Observables?
• Would scaling them back a great deal reduce the size of the library significantly?

Other things

Having schedulers might enable us to centralize error handling in the next version, which could help performance and reduce library size. This was one of the central changes in "T-Rex".

NOTE: This is just a discussion to get us to question what we've been doing with schedulers, not a change suggestion per say

[staltz]

For a slightly different perspective on testing, I invite you to take a look at cycle/time, a tool that @Widdershin built inspired by the RxJS TestScheduler: https://github.com/cyclejs/cyclejs/tree/master/time#rxjs-why-would-i-want-to-use-cycletime-over-the-testscheduler-from-rxjs

The difference there is that the time-related operator, like delay, is passed as an input, as opposed to the scheduler passed as an input and used in the time-related operator.

Globals are tempting, but they are magic and sources of confusion. In Cycle.js we barely have any globals, but the ones we do have, such as the innocently small setAdapt, are constantly a source of discussion/annoyance.

I think the story around schedulers isn't entirely about testing, though. Issues like this combineLatest quirk are real gotchas for beginners and still a source of pain when migrating from RxJS v4 to v5+.

[benlesh]

I think the gotcha for beginners in the case outlined is really around not understanding that Observable.of is a synchronous loop that emits values.

Another thing to think about is the fact that our Observable.of and Observable.from are not really compatible with the current TC39 proposal. It would be nice to bring that in line.

In practice, I really do think it's worth taking a step back and asking what the schedulers really provide.

Globals are tempting, but they are magic and sources of confusion.

As for this, right now all of our provided scheduler instances are essentially global. They are a single, shared, stateful instance at a top level. They might not live in window or global but they are nonetheless accessible from almost anywhere and contain some shared state.

I think the larger problems are:

1. They're rarely needed
2. When one is needed, you have to pass it all over the place to every Observable creation method and timing operator.
3. In most cases, a separate Observable creation method could be provided that would eliminate the need for a scheduler argument.
4. I'm unsure it's good practice to have a method where one argument completely changes the behavior of the resulting type. Observable.of(1) vs Observable.of(1, async). For one thing, it complicates the method, for another thing, it means results are unpredictable for Observable.of(x, y) where you don't know what y is. (Yay dynamic languages).

[staltz]

In practice, I really do think it's worth taking a step back and asking what the schedulers really provide.

Simply put, schedulers solve concurrency ambiguities in a declarative style. The real problem of that Observable.of example is not its synchronicity, but the fact that combineLatest is sensitive to the ordering of input Observables: combineLatest(a$, b$) or combineLatest(b$, a$).

Although Schedulers were originally introduced to Rx(.NET) and heavily important for thread management, concurrency also exists in "single-threaded" JavaScript. Besides the simple Observable.of examples, concurrency problems with reactive streams occur in less obvious cases. For instance, race conditions that we found in Cycle.js. These conditions emerge easily once you have a multicasted observable and you need multiple observers subscribed and not have them compete in a race.

A simple statement of the important problem that schedulers solve: "how can we subscribe to both a$ and b$ simultaneously?". You cannot reliably provide an answer to this if the only thing that exists is the recursive scheduler. Schedulers allow you to describe what does "simultaneously" mean case by case.

I understand that testing is clumsy, we can explore our options for that. I understand that we could simplify the codebase by deleting some parts. But let's not reinvent the wheel or throw away an important part of RxJS, which also makes it uniquely interesting over alternatives like Most.js, xstream or Bacon.js.

As a secondary note, I much prefer Observable.asyncOf(1, 2, 3) over Observable.of(1, 2, 3, scheduler).

[benlesh]

Simply put, schedulers solve concurrency ambiguities in a declarative style.

There is very little that is declarative about how Rx is using schedulers. It's pretty imperative... callThis(arguments, here, btw, scheduler). You can't move the scheduler around to change the behavior, it's a pretty specific instruction to the method you're calling. "create this observable, have it use this scheduler".

A more declarative solution would be specialized operators similar to subscribeOn or observeOn...

source$.async()

// or 

source$.queue()

This would mean your combineLatest problem would be solved like a$.queue().combineLatest(b$.queue(), (a, b) => a + b). So that on the rare and strange occasion you're dealing with all synchronous observables that you want to deal with in a breadth-first manner, you can do so.

concurrency also exists in "single-threaded" JavaScript.

This is the entire basis of Rx. So I know this.

Currently dealing with scheduler defaults in operators is a huge problem when trying to create a test suite. If you're trying to test code that is using .delay() or the like, or is even creating Observables with something like interval or even range internally, and doesn't allow you to pass a scheduler around to everything, the test suite is screwed. That's not a big win.

Scheduling seems to be something that was designed to accommodate 100% of the use cases 100% of the time, but in practice some of what it accommodates doesn't make sense. "I want immediate scheduling on a$, but animationFrame scheduling on b$ and the delay operation I'd like to be scheduled virtually". I mean, it could happen... but is it worth the additional download size and code complexity? That's the real question I'm asking.

an important part of RxJS, which also makes it uniquely interesting over alternatives like Most.js, xstream or Bacon.js.

It also makes it uniquely heavier than any of those mentioned.

So maybe better questions are:

1. How much do schedulers and supporting schedulers cost in terms of size and performance?
2. Do they live up to their value proposition in JavaScript-land given the cost of 1 above?

I suspect that some members of our core team has a very skewed point of view compared to the to outside developers who have started using RxJS in droves over the last year. The MO for most of us, myself included, has been resistance to change and reluctance to explore alternatives.

[benlesh]

If we could simplify the library, make it smaller and faster, and have deterministic tests and still support composition of different scheduling behaviors, is that something worth exploring?

[benlesh]

ALSO: I'm playing devil's advocate this is exploratory, I'm taking the "what if" stance to contrast other opinions and keep people thinking.

Just to be clear.. I'm not arguing with @staltz ... just want to get people thinking.

[benlesh]

So one common use case for me and schedulers is for animations like:

Observable.interval(0, animationFrame)

// or

Observable.range(0, Number.POSITIVE_INFINITY, animationFrame)

// or

Observable.timer(0, 0, animationFrame)

But I wonder if it would be better/more efficient/easier to understand/etc to just have:

Observable.animationFrames(); 

[kwonoj]

subscribeOn / observeOn is not available usecase in here?

[staltz]

I still maintain my previous arguments, but after having thought about this for some hours I realized it's enough (for the JavaScript Rx) to be able to specify breadth-first ordering or depth-first ordering. I don't see the benefit of multiple Schedulers that accomplish breadth-first, we could have just one breadth-first Scheduler. So there's some opportunity for code deletion somewhere.

[staltz]

I noticed yet another cool and unexplored benefit of a breadth-first scheduler: avoiding ReplaySubject or BehaviorSubject when subscribing to a multicasted observable.

Note

const a$ = Rx.Observable.interval(1000).startWith(-1)
  .publishReplay(1).refCount();

a$.subscribe(x => console.log(x));
a$.subscribe(x => console.log(x));

versus

const a$ = Rx.Observable.interval(1000).startWith(-1)
  .observeOn(Rx.Scheduler.asap)
  .publish().refCount();

a$.subscribe(x => console.log(x));
a$.subscribe(x => console.log(x));

[trxcllnt]

@benlesh I'm still in favor of an option to specify the scheduler at subscription time, which would compose all the way through the operator subscriptions and could be used in place of operator defaults, e.g. subscribe(observer, scheduler). Tests would then just need to pass the TestScheduler as the second argument when they subscribe.

[staltz]

Great idea, Paul. Are there any obstacles for doing that? And I suppose observeOn would interrupt that bottom-up propagation.

Since the recursive scheduler is basically "no scheduler", how would the default case work, given subscribe(observer) and supposing we have a delay(1000) upstream? Would the delay just default to asap like it does today?

[trxcllnt]

@staltz yeah operators would still use the scheduler passed in at Observable creation time, to use when there isn't a Subscriber scheduler. To be clear I haven't thought through all the implications of allowing the Subscriber to control producers this way. For instance some funny things might happen inside operators like expand.

Also since the TestScheduler trampolines like the QueueScheduler, we'd need to have a RecursiveTestScheduler for people who want to test the normal recursive behavior.

[davidwdan]

With earlier versions of RxPHP we specified the scheduler at subscription time. The main problem we ran into was when you implemented a new operator, you had to remember to pass the scheduler through to the inner subscription. If you didn't, the scheduler would stop propagating upstream, which led to bugs that were very difficult to debug.

[MVSICA-FICTA]

Scheduling seems to be the occult side of RxJS. I was at first drawn toward RxJS Schedulers because the similar workings (queue, context, clock) they shared with the music schedulers presented by IRCAM, where a robust scheduling foundation was extended to many usage cases.

Presentation: https://medias.ircam.fr/x6c8804
Submission: https://wac.ircam.fr/pdf/wac15_submission_19.pdf

Looking at the RxJS Scheduler implementation it almost seemed that they could be made to work for music. Just like there is a RAF Scheduler there could also be a WAA Scheduler (for the cross browser Web Audio API).

Maybe it would be possible to schedule each note from a parsed score using the currentTime of the WAA context/thread? However, this becomes a bit more involved than just scheduling against a periodic pulse like RAF does. The periodic pulse is a macro level of scheduling where the note-by-note is a micro level of scheduling. Also, this is different than clocking external MIDI signals, where the macro level would suffice.

Anytime I've looked into RxJS scheduling the focus tends to be about the Virtual Scheduler and macro level scheduling. At this junction experience tells me that there are other ways to schedule music that involve a combo of RxJS and some quintessential JavaScript.

Earlier, I had posted a question about music scheduling to @staltz (a musician) at the RxJS AMA he hosted at HashNode, but that went nowhere after 265 views and not one reply. Below is a link to the post just so it can be logged into the journal of RxJS time. I still hope for a future where all the expressions of time (music, animation, async) can be united in a common RxJS foundation.

https://hashnode.com/post/is-a-custom-rxjs-scheduler-required-for-accurate-music-playback-cj5vu9hs600zmjpwtgd2fztwp

[benlesh]

@MVSICA-FICTA ... I've looked at web audio scheduiing for Observables, and I came to the same conclusion I have with the requestAnimationFrame stuff, most of the time what you want is an observable of moments, not really a scheduler. And the former is much easier to engineer.

The overall issue we have with Schedulers right now in RxJS, IMO, is that we've over-engineered the solution a bit. Scheduling makes more sense in a multi-threaded environment. The only scheduling distinction I've seen make a lot of sense for us is breadth-first vs depth-first scheduling, but even that is fairly edge-casey, honestly.

What do I mean by over-engineered? Well, anything with any sort of delay really just ends up using setTimeout unless there's some virtual scheduling going on, which is really only for testing purposes for the most part. No matter what scheduler you use, if you schedule with a delay of more than zero, it's going to be a setTimeout. I think that the animationFrame scheduler might be an exception there, but honestly, I'm still not sure animationFrame is best as a scheduler, when it could just be an observable.

If we stepped back to use wrapped, native scheduling APIs, we could reduce a lot of code, and simplify and even speed up our operators substantially. Not to mention step away from what's really one of the most imperative parts of RxJS, which is scheduler specification. (Which again, very very few people use or understand)

[jhusain]

Sounds like one of the key concerns here is the additional file size of overloaded operators that accept optional schedulers. Why not split these operators in two? For example have Observable.of, and Observable.ofWithScheduler? This way developers who are not using schedulers don’t have to pay for them.

Alternately if we can find a way to dramatically reduce the file size by simplifying the available schedulers, that would be nice as well.

[trxcllnt]

@benlesh The overall issue we have with Schedulers right now in RxJS, IMO, is that we've over-engineered the solution a bit. Scheduling makes more sense in a multi-threaded environment. The only scheduling distinction I've seen make a lot of sense for us is breadth-first vs depth-first scheduling, but even that is fairly edge-casey, honestly.

Scheduling is about concurrency, not threading. Threads are a type of concurrency, but not the only kind. That said with SharedArrayBuffers and workers, JS is getting a type of threading whether we like it or not.

Well, anything with any sort of delay really just ends up using setTimeout

Do you mean JS in general, or in Rx? We explicitly use setInterval instead of setTimeout in the AsyncScheduler, because setTimeout can't reliably execute intervals (esp. under load), nor does it reliably synchronize with items scheduled using setInterval, leading to heisenbugs.

I think that the animationFrame scheduler might be an exception there, but honestly, I'm still not sure animationFrame is best as a scheduler, when it could just be an observable.

Synchronizing events on the animationFrame is an absolute necessity for performance. It's totally a scheduler thing. events.auditTime(0, Scheduler.animationFrame).subscribe(render) is one of the most common things in our front-end code.

If we stepped back to use wrapped, native scheduling APIs, we could reduce a lot of code, and simplify and even speed up our operators substantially.

First, we do use the native scheduling APIs. Schedulers aren't meant to replace those APIs, they're meant to abstract calling those APIs away from the specific Observable/operator implementations. By abstracting the logic of calling native scheduling APIs, we can then parameterize which APIs Observables and operators ultimately use. This is a critical component of what makes Rx a superset of pure FRP.

Second, the schedulers are very lean. The current design was meant to simplify/unify the scheduling interface behind a single schedule method that can schedule either one or infinite executions, with or without state, for fixed or changing periods. The API was explicitly designed so we could eliminate additional Subscription allocations across intervals by implementing a form of async tail-recursion, which improves scheduler performance and reduces GC churn:

// This only allocates a single Subscription for infinite execution
scheduler.schedule(function(i) {
  this.schedule(i+1, 100);
}, 0, 100);

Considering the apparent difficulty of implementing efficient/safe async scheduling, I'm extremely skeptical of the safety of hard-coding scheduling logic into the Observables and operators themselves. Seems like a heap 'o trouble.

@jhusain Why not split these operators in two? For example have Observable.of, and Observable.ofWithScheduler? This way developers who are not using schedulers don’t have to pay for them.

What do you mean? Unless something's changed since I last looked, of doesn't have a default scheduler.

Alternately if we can find a way to dramatically reduce the file size by simplifying the available schedulers, that would be nice as well.

I've mentioned before to @benlesh we absolutely can reduce the LOC in the Schedulers. Frankly the only reason they're broken down into so many classes is because I didn't want to confuse anybody in the last big scheduler PR. I'd be happy to submit a PR that reduces the LOC in schedulers by combining logic into a single parameterized Action class.

[benlesh]

Do you mean JS in general, or in Rx?

I was generalizing. I remember that we're using setInterval.

What do you mean? Unless something's changed since I last looked, of doesn't have a default scheduler.

It's more that you can provide a scheduler, and there's additional code inside of of to accommodate that. This is a pattern that exists throughout the library. By moving scheduled versions of operators to their own implementation, the common-use-case operators that don't have any scheduling will become much lighter and simpler. Likewise, in operators like delay, timer or interval, I think we could just use setInterval or setTimeout in a more direct fashion. There are ways to have deterministic tests for those sorts of things. It almost never makes sense to provide a queue scheduler or an asap scheduler to timer or interval... or even delay (that's what observeOn is for).

At this point, I think we should keep scheduling, but isolate it from all of the common use cases. Most things can be handled with subscribeOn or observeOn.

[staltz]

Agreed with the above, except for direct use of setInterval in delay (and others) operators. Because that would mean we depend on a global, and for testing we mutate the global. Would introduce a new class of bugs when testing.

[trxcllnt]

@benlesh It's more that you can provide a scheduler, and there's additional code inside of of to accommodate that. This is a pattern that exists throughout the library. By moving scheduled versions of operators to their own implementation, the common-use-case operators that don't have any scheduling will become much lighter and simpler.

Totally agree with making things simpler, but I wouldn't be surprised if we couldn't accomplish the same thing with a helper function that handles the scheduler/no scheduler case generically. The duplication in this area is largely a product of multiple cooks in the kitchen, not a fundamentally unsound abstraction.

Likewise, in operators like delay, timer or interval, I think we could just use setInterval or setTimeout in a more direct fashion.

👎 from me on this, I frequently use timer and interval with schedulers other than the AsyncScheduler.

It almost never makes sense to provide a queue scheduler or an asap scheduler to timer or interval... or even delay (that's what observeOn is for).

observeOn ultimately losslessly buffers events using the scheduler as the buffer queue, which is absolutely not desired in many cases. For example range(0, 100000, Scheduler.asap) will efficiently emit 100k numbers over 100k macrotasks without buffering. range(0, 100000).observeOn(Scheduler.asap) will immediately buffer 100k notifications as AsapActions into the AsapScheduler queue, which will all be flushed synchronously on the next macrotask. Not the same thing at all.

[paulpdaniels]

I'm personally a fan of what @trxcllnt mentioned above, the idea of passing in schedulers at (or closer to) subscription time make a whole bunch more sense to me than declaring them at construction time. It also gives a much purer taste to the end product. For my own attempts to build a streaming library I looked at doing an interface like:

const delayTime = (delayAmount) => (source, scheduler) => /*Apply delay operator*/.

The ergonomics were still a bit rough and there were some issues around passing schedulers and using defaults, but it does allow for some interesting combinatorial setups, such as applying a scheduler as context. Something along the lines of:

const unscheduled =
  pipe(
    delayTime(1000),
    bufferCount(10)
  );

// Overrides any incoming scheduler with async
const scheduled = scheduler.async.with(unscheduled);

// Applies a default scheduler which is overridden downstream
scheduled(Observable.fromEvent(window, 'keypress'), scheduler.default)
  .subscribe(observer)

Again interface is a bit kludgy, but it pulls the scheduling away from the operator so they each exist in their own layer and makes it easier to test things since there isn't a reliance on a global scheduler entity.

[Widdershin]

Currently dealing with scheduler defaults in operators is a huge problem when trying to create a test suite. If you're trying to test code that is using .delay() or the like, or is even creating Observables with something like interval or even range internally, and doesn't allow you to pass a scheduler around to everything, the test suite is screwed. That's not a big win.

This is a solved problem with @cycle/time.

Instead of taking schedulers as arguments, we create a scheduler which has time based operators.

So instead of:

import {Observable, TestScheduler} from 'rxjs';
import * as assert from 'assert';

function main(scheduler = null) { // does this even work or do I need to set the correct default?
  return Observable.of('woo').delay(60, scheduler);
}

// in prod
const result = main();

// in test
const testScheduler = new TestScheduler(assert.deepEqual.bind(assert));
const result = main(testScheduler);

const expected = '--x';
const expectedStateMap = {x: 'woo'};

testScheduler.expectObservable(result).toBe(expected, expectedStateMap);

testScheduler.flush();

Using @cycle/time with RxJS, it would look like this:

import {Observable} from 'rxjs';
import {timeDriver, mockTimeSource} from '@cycle/time/rxjs';

function main(Time = timeDriver()) {
  return Observable.of('woo').let(Time.delay(60));
}

// in prod
const result = main();

// in test
const Time = mockTimeSource();
const result = main(Time);

const expected = Time.diagram('--x', {x: 'woo'});

Time.assertEqual(result, expected);

Time.run();

They have very similar APIs, but Time sources have delay, debounce, periodic, throttle and the capacity to make new time based operators.

Some will dislike how @cycle/time encourages the use of dependency injection, preferring a global singleton to avoid needing to pass arguments around.

This is also easily achievable with @cycle/time. We can create a module that instantiates a timeDriver() and exports it, then import it anywhere needed. In test, we would mock out the global and instead use a mockTimeSource.

But I wonder if it would be better/more efficient/easier to understand/etc to just have:
Observable.animationFrames();

This is the approach that @cycle/time takes. @cycle/time implements Time.animationFrames(), and also an operator for limiting events to one per frame, throttleAnimation(). In tests, these operators are still provided and the animation frame semantics are mocked out.

@cycle/time is also compatible with any Observable symbol libraries, and explicitly supports xstream, rxjs and most.js.

Pros:

• Don't need to remember to pass scheduler as argument
• Handles requestAnimationFrame usecases in a user friendly way
• Encourages dependency injection, allows global approach if desired
• Writing tests does not require changing any app code
• Personally, I think @cycle/time has a friendlier API than the TestScheduler

Cons:

• Does not have full breadth of rxjs time based operators currently implemented
• Forced to use @cycle/time operators, cannot mix and match

I like the idea of passing the scheduler on subscription, hadn't considered that.

Anyway, just another option. It's currently feasible to use @cycle/time for all your scheduling needs when writing RxJS code. For the curious, I'd recommend checking out the README.

[luchillo17]

Hi there, little question, i was trying to sync a video with a custom timer component, and some times i had to pause 1 of them to get them in sync, seems the Rxjs timer is not precise enough and when my update interval is small (say 10 milliseconds) they get off sync very easily.

I was reading this article regarding such thing and i'm still a bit lost about how to fix this: Is a custom rxjs scheduler required for accurate music playback?

I think it's kind of related to the animation frame discussed above in some comments.

[brucou]

Great discussion. I am adding a few remarks, as seen from an API user perspective :

• It's a little-understood part of the library : well yes. But that seems easy to fix by improving the documentation. The only way I got to understand the subject is by going to the code, as the present documentation was a bit short (talking about rxjs v4 here, rxjs v5 doc is so much better). Bread and depth scheduling are definitely worth their own sections. I can help with that for v4, which is what I know extensively about.

• what set aside rxjs for me is precisely that is has schedulers, and that the API consumer can act on them when needed. I do get the argument about size of the library, testing, Observable standard proposal etc. Still, accommodate 100% of the use cases 100% of the time, is what makes me stick to Rxjs, after having been burnt with other streaming libraries. Stuff like publish and connect for instance is a part of this 100% guarantee and for me that is invaluable. I even got to use singleInstance (even though only once). Bottomline is I know that when the defaults go wrong, I can tweak it.

• at the core of this issue I believe is how you conceptualize streams. If you see them as a list of values, which some users tend to do (say a kind of LazyList), then schedulers might be of anecdotical value (time matters only on the edge, and testing is significantly simplified, you just check that the expected values are received in order). If you see them as values with time information attached (which I do), then schedulers may help to enforce timing properties, and on the flip side. testing is complicated because you have to test that values arrive in sequence but also in a timely manner.

• if you see streams as endowed with time information, then I think it is hard to give a coherent meaning to $.of(1,2) without any further information. In the current implementation, I would denote it by [(1, t0), (2, t0)] which is confusing : a given stream observed at any given instant should only have one value. At least, it makes it much easier to reason about and build higher abstractions on top of it. There was this operator generateWithAbsoluteTime in rxjs v4 which exemplifies that you may want to emit a sequence of values, but also specify the timing of the sequence too. In short the problem with Observable.of(x, y) is that y is on the same time as x in the same observable, and yet Observable.of(x, y) has a different meaning from Observable.of(y, x), as observed from streams derived from it (cf. comment and example here : combineLatest on synchronous sources #2414). It does not matter if the first value is synchronous or not, but having subsequent values in another time gives stronger guarantees for reasoning. So either that 'another' time is specified explicitly as in generateWithAbsoluteTime, or it is specified implicitly through a scheduler. This would eliminate the impredictability that Ben mentions for Observable.of(x, y).

• lastly, @benlesh suggests a$.queue().combineLatest(b$.queue(), (a, b) => a + b) to deal with non-commutativity of combineLatest in the case of both a and b being synchronous streams. However, from the point of view of somebody building higher abstractions on top of rx, you do not always have readily available that information of synchronicity/asynchronicity about the streams you manipulate.

[kievsash]

👎 from me on this, I frequently use timer and interval with schedulers other than the AsyncScheduler.

Sorry for interference, but can you provide use-cases when it is needed?

[kirk-l]

Why doesn't the animationFrames Observable -- added in #5021 -- pass the DOMHighResTimeStamp that the requestAnimationFrames callback receives, to its subscribers? According to MDN:

When callbacks queued by requestAnimationFrame() begin to fire multiple callbacks in a single frame, each receives the same timestamp even though time has passed during the computation of every previous callback's workload.

This is critical when running multiple rAF loops to keep them in sync. The current default implementation, using Date.now(), will emit slightly different values for each loop and to each subscriber.

Here's the implementation I use:

const animationFrames = new Observable(subscriber => {
    let requestId = window.requestAnimationFrame(function step(timestamp) {
        subscriber.next(timestamp);
        if (!subscriber.closed) {
            requestId = window.requestAnimationFrame(step);
        }
    });
    return () => window.cancelAnimationFrame(requestId);
});

And to get the elapsed milliseconds:

const elapsedMs = defer(() => {
    let start = null;
    return animationFrames.pipe(
        tap(timestamp => start === null && (start = timestamp)),
        map(timestamp => timestamp - start),
    );
});

I think this would be a better animationFrames Observable.

[cartant]

@kirk-l Could you open a new issue for what you've raised in the above comment? I'd suggest opening it as a bug, as the current observable API doesn't allow usage like what's possible with the rAF API.

[joshribakoff]

Schedulers seem to overlap with what operators can already do, both can manipulate the stream with the dimension of time. I don't see what the benefit is to being able to pass the scheduler to the operators themselves, vs just adding an .observeOn(scheduler) before/after any operator in the pipe() call.

In either case, it's really important there is some notion of being able to change the execution context. For example, I'm planning on exploring a React RxJS scheduler (rx-store/rx-store#13). Eg - if numerous subscriptions are to be delivered a "next" value in any given upcoming animation frame, you'd have the option to ensure all of these subscriptions have their value delivered in some execution context such as a React batched update callback. This is similar to what @kirk-l points out, but this isn't specific to animation frames fyi @cartant . There is a valid use case to being able to change the execution context for subscribing and observing, in general. Painting a bunch of React updates in a singe animation frame, in a single batched update, for example, even when they come from separate subscriptions.

Really though this all seems like it could be done with operators. An operator receiving values synchronously could just next them to the output observable asynchronously. You'd still have to "pass the operator all around" instead of importing it, to solve the testing concern, but then again I don't think RxJS should try be a DI container. Having the option to just keep the setTimeout() call & use something like Sinon or Jest fake timers would be nice, it is technically possible now, too, just not widely understood.

Another thing to consider is if your source of time could be remote, or could come from external user input (like a timeline component to scrub through a powerpoint presentation style app, triggering animations in virtual time controlled by a slider on a timeline UI component), but again maybe scheduler is redundant with observable/operator abstraction here. The user can just swap out an observable of the current time with an observable of animation frame timestamps or Date.now() timestamps or performance.now() timestamps, and maybe the separate scheduler abstraction isn't strictly needed in its current from.

Maybe scheduler() can be more of its own operator, vs something all operators are aware of? Actually this won't work, if you want the delay() operator to obey the virtual time controlled by the timeline slider UI component in the above example, then delay() (and all other time based operators need a scheduler, because you cant really use jest/sinon for production code that needs virtual time)

@staltz suggestion of having a steam of time is not ideal, either. Ideally I do not need a stream that emits every 1ms going on all the time, in the name of "engineering purity". Ideally, my code written in RxJS should not be calling performance.now() unless there's actually strictly a need to do so.