feat(startWithTimeout): Add operator.

source/operators/index.ts

@@ -18,6 +18,7 @@ export * from "./rateLimit";
 export * from "./refCountAuditTime";
 export * from "./refCountOn";
 export * from "./reschedule";
+export * from "./startWithTimeout";
 export * from "./subsequent";
 export * from "./switchMapUntil";
 export * from "./takeWhileInclusive";

source/operators/startWithTimeout-spec.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
+ */
+/*tslint:disable:no-unused-expression*/
+
+import { startWithTimeout } from "./startWithTimeout";
+import { marbles } from "rxjs-marbles";
+
+describe("startWithTimeout", () => {
+
+    it("should do nothing if the source emits within the duration", marbles(m => {
+
+        const source =   m.cold("--a--b--c--|");
+        const subs =            "^----------!";
+        const expected = m.cold("--a--b--c--|");
+
+        const destination = source.pipe(startWithTimeout("z", m.time("---|")));
+        m.expect(destination).toBeObservable(expected);
+        m.expect(source).toHaveSubscriptions(subs);
+    }));
+
+    it("should emit the value if the source is empty", marbles(m => {
+
+        const source =   m.cold("-----------|");
+        const subs =            "^----------!";
+        const expected = m.cold("---z-------|");
+
+        const destination = source.pipe(startWithTimeout("z", m.time("---|")));
+        m.expect(destination).toBeObservable(expected);
+        m.expect(source).toHaveSubscriptions(subs);
+    }));
+
+    it("should emit the value if the source emits after the duration", marbles(m => {
+
+        const source =   m.cold("-----a-b-c-|");
+        const subs =            "^----------!";
+        const expected = m.cold("---z-a-b-c-|");
+
+        const destination = source.pipe(startWithTimeout("z", m.time("---|")));
+        m.expect(destination).toBeObservable(expected);
+        m.expect(source).toHaveSubscriptions(subs);
+    }));
+});

source/operators/startWithTimeout.ts

@@ -0,0 +1,14 @@
+/**
+ * @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 { concat, MonoTypeOperatorFunction, race, SchedulerLike, timer } from "rxjs";
+import { mapTo, publish } from "rxjs/operators";
+
+export function startWithTimeout<T>(value: T, duration: number | Date, scheduler?: SchedulerLike): MonoTypeOperatorFunction<T> {
+    return source => source.pipe(publish(published => race(
+        published,
+        concat(timer(duration, scheduler).pipe(mapTo(value)), published)
+    )));
+}