feat(multicast): add higher-order lettable variant of multicast

Also adds MonoTypeOperatorFunction

Author: benlesh

src/interfaces.ts

@@ -3,3 +3,7 @@ import { Observable } from './Observable';
 export type UnaryFunction<T, R> = (source: T) => R;
 
 export type OperatorFunction<T, R> = UnaryFunction<Observable<T>, Observable<R>>;
+
+export type FactoryOrValue<T> = T | (() => T);
+
+export type MonoTypeOperatorFunction<T> = OperatorFunction<T, T>;

src/operator/multicast.ts

@@ -1,12 +1,12 @@
 import { Subject } from '../Subject';
-import { Operator } from '../Operator';
-import { Subscriber } from '../Subscriber';
 import { Observable } from '../Observable';
-import { ConnectableObservable, connectableObservableDescriptor } from '../observable/ConnectableObservable';
+import { ConnectableObservable } from '../observable/ConnectableObservable';
+import { multicast as higherOrder } from '../operators';
+import { FactoryOrValue, MonoTypeOperatorFunction } from '../interfaces';
 
 /* tslint:disable:max-line-length */
-export function multicast<T>(this: Observable<T>, subjectOrSubjectFactory: factoryOrValue<Subject<T>>): ConnectableObservable<T>;
-export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: selector<T>): Observable<T>;
+export function multicast<T>(this: Observable<T>, subjectOrSubjectFactory: FactoryOrValue<Subject<T>>): ConnectableObservable<T>;
+export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: MonoTypeOperatorFunction<T>): Observable<T>;
 /* tslint:enable:max-line-length */
 
 /**
@@ -30,38 +30,5 @@ export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>
  */
 export function multicast<T>(this: Observable<T>, subjectOrSubjectFactory: Subject<T> | (() => Subject<T>),
                              selector?: (source: Observable<T>) => Observable<T>): Observable<T> | ConnectableObservable<T> {
-  let subjectFactory: () => Subject<T>;
-  if (typeof subjectOrSubjectFactory === 'function') {
-    subjectFactory = <() => Subject<T>>subjectOrSubjectFactory;
-  } else {
-    subjectFactory = function subjectFactory() {
-      return <Subject<T>>subjectOrSubjectFactory;
-    };
-  }
-
-  if (typeof selector === 'function') {
-    return this.lift(new MulticastOperator(subjectFactory, selector));
-  }
-
-  const connectable: any = Object.create(this, connectableObservableDescriptor);
-  connectable.source = this;
-  connectable.subjectFactory = subjectFactory;
-
-  return <ConnectableObservable<T>> connectable;
-}
-
-export type factoryOrValue<T> = T | (() => T);
-export type selector<T> = (source: Observable<T>) => Observable<T>;
-
-export class MulticastOperator<T> implements Operator<T, T> {
-  constructor(private subjectFactory: () => Subject<T>,
-              private selector: (source: Observable<T>) => Observable<T>) {
-  }
-  call(subscriber: Subscriber<T>, source: any): any {
-    const { selector } = this;
-    const subject = this.subjectFactory();
-    const subscription = selector(subject).subscribe(subscriber);
-    subscription.add(source.subscribe(subject));
-    return subscription;
-  }
+  return higherOrder(<any>subjectOrSubjectFactory, selector)(this);
 }

src/operators/index.ts

@@ -5,6 +5,7 @@ export { map } from './map';
 export { max } from './max';
 export { mergeMap } from './mergeMap';
 export { min } from './min';
+export { multicast } from './multicast';
 export { reduce } from './reduce';
 export { scan } from './scan';
 export { switchMap } from './switchMap';

src/operators/multicast.ts

@@ -0,0 +1,67 @@
+import { Subject } from '../Subject';
+import { Operator } from '../Operator';
+import { Subscriber } from '../Subscriber';
+import { Observable } from '../Observable';
+import { ConnectableObservable, connectableObservableDescriptor } from '../observable/ConnectableObservable';
+import { FactoryOrValue, MonoTypeOperatorFunction } from '../interfaces';
+
+/* tslint:disable:max-line-length */
+export function multicast<T>(subjectOrSubjectFactory: FactoryOrValue<Subject<T>>): MonoTypeOperatorFunction<T>;
+export function multicast<T>(SubjectFactory: (this: Observable<T>) => Subject<T>, selector?: MonoTypeOperatorFunction<T>): MonoTypeOperatorFunction<T>;
+/* tslint:enable:max-line-length */
+
+/**
+ * Returns an Observable that emits the results of invoking a specified selector on items
+ * emitted by a ConnectableObservable that shares a single subscription to the underlying stream.
+ *
+ * <img src="./img/multicast.png" width="100%">
+ *
+ * @param {Function|Subject} subjectOrSubjectFactory - Factory function to create an intermediate subject through
+ * which the source sequence's elements will be multicast to the selector function
+ * or Subject to push source elements into.
+ * @param {Function} [selector] - Optional selector function that can use the multicasted source stream
+ * as many times as needed, without causing multiple subscriptions to the source stream.
+ * Subscribers to the given source will receive all notifications of the source from the
+ * time of the subscription forward.
+ * @return {Observable} An Observable that emits the results of invoking the selector
+ * on the items emitted by a `ConnectableObservable` that shares a single subscription to
+ * the underlying stream.
+ * @method multicast
+ * @owner Observable
+ */
+export function multicast<T>(subjectOrSubjectFactory: Subject<T> | (() => Subject<T>),
+                             selector?: (source: Observable<T>) => Observable<T>): MonoTypeOperatorFunction<T> {
+  return function multicastOperatorFunction(source: Observable<T>): Observable<T> {
+    let subjectFactory: () => Subject<T>;
+    if (typeof subjectOrSubjectFactory === 'function') {
+      subjectFactory = <() => Subject<T>>subjectOrSubjectFactory;
+    } else {
+      subjectFactory = function subjectFactory() {
+        return <Subject<T>>subjectOrSubjectFactory;
+      };
+    }
+
+    if (typeof selector === 'function') {
+      return source.lift(new MulticastOperator(subjectFactory, selector));
+    }
+
+    const connectable: any = Object.create(source, connectableObservableDescriptor);
+    connectable.source = source;
+    connectable.subjectFactory = subjectFactory;
+
+    return <ConnectableObservable<T>> connectable;
+  };
+}
+
+export class MulticastOperator<T> implements Operator<T, T> {
+  constructor(private subjectFactory: () => Subject<T>,
+              private selector: (source: Observable<T>) => Observable<T>) {
+  }
+  call(subscriber: Subscriber<T>, source: any): any {
+    const { selector } = this;
+    const subject = this.subjectFactory();
+    const subscription = selector(subject).subscribe(subscriber);
+    subscription.add(source.subscribe(subject));
+    return subscription;
+  }
+}