feat(continueWith): Add continueWith (#90)

source/operators/continueWith-spec.ts

@@ -0,0 +1,45 @@
+/**
+ * @license Use of this source code is governed by an MIT-style license that
+ * can be found in the LICENSE file at https://github.com/cartant/rxjs-etc
+ */
+/*tslint:disable:no-unused-expression*/
+
+import { of } from "rxjs";
+import { marbles } from "rxjs-marbles";
+import { continueWith } from "./continueWith";
+
+// prettier-ignore
+describe("continueWith", () => {
+  it(
+    "should continue with the latest emitted value",
+    marbles(m => {
+      const source = m.cold(" abc|");
+      const expected = "      abc(c|)";
+
+      const result = source.pipe(continueWith((x) => of(x)));
+      m.expect(result).toBeObservable(expected);
+    })
+  );
+
+  it(
+    "should not continue if the source has not emitted",
+    marbles(m => {
+      const source = m.cold(" ---|");
+      const expected = "      ---|";
+
+      const result = source.pipe(continueWith((x) => of(x)));
+      m.expect(result).toBeObservable(expected);
+    })
+  );
+
+  it(
+    "should not continue if the mapped observable is EMPTY",
+    marbles(m => {
+      const source = m.cold(" abc|");
+      const expected = "      abc|";
+
+      const result = source.pipe(continueWith(() => []));
+      m.expect(result).toBeObservable(expected);
+    })
+  );
+});

source/operators/continueWith.ts

@@ -0,0 +1,44 @@
+/**
+ * @license Use of this source code is governed by an MIT-style license that
+ * can be found in the LICENSE file at https://github.com/cartant/rxjs-etc
+ */
+
+import {
+  from,
+  Observable,
+  ObservableInput,
+  ObservedValueOf,
+  OperatorFunction,
+  Subscription,
+} from "rxjs";
+
+const NO_VAL: any = {};
+
+export const continueWith = <T, O extends ObservableInput<any>>(
+  mapper: (value: T) => O
+): OperatorFunction<T, T | ObservedValueOf<O>> => (source$) =>
+  new Observable((observer) => {
+    let latestValue: T = NO_VAL;
+    const subscription = new Subscription();
+
+    subscription.add(
+      source$.subscribe({
+        next: (val) => {
+          observer.next((latestValue = val));
+        },
+        error: (e) => {
+          observer.error(e);
+        },
+        complete: () => {
+          if (latestValue === NO_VAL) {
+            observer.complete();
+          } else {
+            const nextObservable$ = from(mapper(latestValue));
+            subscription.add(nextObservable$.subscribe(observer));
+          }
+        },
+      })
+    );
+
+    return subscription;
+  });