feat: add scheduler for Run Loop

addon/index.js

@@ -1,4 +1,5 @@
 export { default as Route } from "./route";
+export { default as scheduler } from "./scheduler";
 
 export { default as subscribe } from "./decorators/subscribe";
 

addon/scheduler.js

@@ -0,0 +1,23 @@
+import { Scheduler, Subscription } from "rxjs";
+import { cancel, later, run } from "@ember/runloop";
+
+class RunLoopAction extends Subscription {
+  constructor(scheduler, work) {
+    super(() => cancel(this.timer));
+
+    this.scheduler = scheduler;
+    this.work = work;
+  }
+
+  schedule(state, delay) {
+    if (delay) {
+      this.timer = later(this, this.work, state, delay);
+    } else {
+      run(this, this.work, state);
+    }
+
+    return this;
+  }
+}
+
+export default new Scheduler(RunLoopAction);

tests/dummy/app/pods/docs/features/scheduler/template.md

@@ -0,0 +1,22 @@
+# Run Loop Scheduler
+
+RxJS has the concept of a [Scheduler][rxjs-scheduler], which is can use to queue work that should happen sometime in the future. Many Observables or Operators that deal with "time" can take accept a Scheduler as an argument if you need to customize how this work is planned.
+
+Ember also has a means for scheduling tasks in the future; the [Run Loop][run-loop]. In order to schedule events from RxJS into the Run Loop, you can leverage the scheduler provided by `ember-rx`.
+
+```javascript
+import { scheduler as runLoopScheduler } from "ember-rx";
+import { interval } from "rxjs";
+
+const observable = interval(1000, runLoopScheduler);
+const subscription = observable.subscribe(val => {
+  // Each emission from `interval` will be scheduled into a Run Loop
+  console.log(value);
+});
+
+// Unsubscribing will cancel any scheduled-but-not-run Run Loops
+subscription.unsubscribe();
+```
+
+[rxjs-scheduler]: https://rxjs.dev/guide/scheduler
+[run-loop]: https://guides.emberjs.com/release/applications/run-loop/

tests/dummy/app/pods/docs/template.hbs

@@ -4,6 +4,9 @@
   {{#viewer.nav as |nav|}}
     {{nav.item 'Introduction' 'docs.index'}}
 
+    {{nav.section 'Features'}}
+    {{nav.item 'Run Loop Scheduler' 'docs.features.scheduler'}}
+
     {{nav.section 'Cookbook'}}
     {{nav.item 'What is this?' 'docs.cookbook.index'}}
     {{nav.item 'Count Property Changes' 'docs.cookbook.count-prop-changes'}}

tests/dummy/app/router.js

@@ -11,6 +11,10 @@ Router.map(function() {
     this.route("cookbook", function() {
       this.route("count-prop-changes");
     });
+
+    this.route("features", function() {
+      this.route("scheduler");
+    });
   });
 
   this.route("testing", function() {

tests/unit/scheduler-test.js

@@ -0,0 +1,66 @@
+import { module, test } from "qunit";
+import td from "testdouble";
+import { getSettledState, settled } from "@ember/test-helpers";
+import { defer } from "rsvp";
+import { from } from "rxjs";
+import { delay } from "rxjs/operators";
+import { scheduler as runloopScheduler } from "ember-rx";
+
+module("Unit | Scheduler", function() {
+  test("it schedules immediate work using the Runloop", async function(assert) {
+    const deferred = defer();
+    const observable = from(deferred.promise, runloopScheduler);
+    const observer = td.function("observer");
+    observable.subscribe(observer);
+
+    deferred.resolve("resolution");
+
+    const state = getSettledState();
+
+    assert.ok(state.hasRunLoop, 'A run loop was started with the "work"');
+
+    assert.verify(
+      observer("resolution"),
+      { times: 0 },
+      "Observer not yet called"
+    );
+
+    await settled();
+
+    assert.verify(
+      observer("resolution"),
+      { times: 1 },
+      "Observer called once the runloop has settled"
+    );
+  });
+
+  module("delayed work", function() {
+    test("it schedules later work using the Runloop", async function(assert) {
+      const observable = from([1]).pipe(delay(10, runloopScheduler));
+      const observer = td.function("observer");
+      observable.subscribe(observer);
+
+      const state = getSettledState();
+
+      assert.ok(
+        state.hasPendingTimers,
+        "Work scheduled for later in the runloop"
+      );
+
+      await settled();
+
+      assert.verify(observer(1), "Eventually called with the scheduled work");
+    });
+
+    test("it can cancel a scheduled runloop", async function(assert) {
+      const observable = from([1]).pipe(delay(10, runloopScheduler));
+      const subscription = observable.subscribe(td.function("observer"));
+
+      subscription.unsubscribe();
+
+      const state = getSettledState();
+
+      assert.notOk(state.hasPendingTimers, "Scheduled work has been cancelled");
+    });
+  });
+});