feat(operators): Use lift in the operators that don't currently use l…

…ift.

BREAKING CHANGE:  Removes MulticastObservable subclass in favor of a MulticastOperator.

spec/Observable-spec.ts

@@ -579,16 +579,22 @@ describe('Observable.create', () => {
 
 /** @test {Observable} */
 describe('Observable.lift', () => {
-  it('should be overrideable in a custom Observable type that composes', (done: MochaDone) => {
-    class MyCustomObservable<T> extends Rx.Observable<T> {
-      lift<R>(operator: Rx.Operator<T, R>): Rx.Observable<R> {
-        const observable = new MyCustomObservable<R>();
-        (<any>observable).source = this;
-        (<any>observable).operator = operator;
-        return observable;
-      }
+
+  class MyCustomObservable<T> extends Rx.Observable<T> {
+    static from<T>(source: any) {
+      const observable = new MyCustomObservable<T>();
+      observable.source = <Rx.Observable<T>> source;
+      return observable;
     }
+    lift<R>(operator: Rx.Operator<T, R>): Rx.Observable<R> {
+      const observable = new MyCustomObservable<R>();
+      (<any>observable).source = this;
+      (<any>observable).operator = operator;
+      return observable;
+    }
+  }
 
+  it('should be overrideable in a custom Observable type that composes', (done: MochaDone) => {
     const result = new MyCustomObservable((observer: Rx.Observer<number>) => {
       observer.next(1);
       observer.next(2);
@@ -610,6 +616,122 @@ describe('Observable.lift', () => {
       });
   });
 
+  it('should compose through multicast and refCount', (done: MochaDone) => {
+    const result = new MyCustomObservable((observer: Rx.Observer<number>) => {
+      observer.next(1);
+      observer.next(2);
+      observer.next(3);
+      observer.complete();
+    })
+    .multicast(() => new Rx.Subject())
+    .refCount()
+    .map((x: number) => { return 10 * x; });
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    const expected = [10, 20, 30];
+
+    result.subscribe(
+      function (x) {
+        expect(x).to.equal(expected.shift());
+      }, (x) => {
+        done(new Error('should not be called'));
+      }, () => {
+        done();
+      });
+  });
+
+  it('should compose through multicast with selector function', (done: MochaDone) => {
+    const result = new MyCustomObservable((observer: Rx.Observer<number>) => {
+      observer.next(1);
+      observer.next(2);
+      observer.next(3);
+      observer.complete();
+    })
+    .multicast(() => new Rx.Subject(), (shared) => shared.map((x: number) => { return 10 * x; }));
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    const expected = [10, 20, 30];
+
+    result.subscribe(
+      function (x) {
+        expect(x).to.equal(expected.shift());
+      }, (x) => {
+        done(new Error('should not be called'));
+      }, () => {
+        done();
+      });
+  });
+
+  it('should compose through combineLatest', () => {
+    const e1 =   cold('-a--b-----c-d-e-|');
+    const e2 =   cold('--1--2-3-4---|   ');
+    const expected = '--A-BC-D-EF-G-H-|';
+
+    const result = MyCustomObservable.from(e1).combineLatest(e2, (a: any, b: any) => String(a) + String(b));
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    expectObservable(result).toBe(expected, {
+      A: 'a1', B: 'b1', C: 'b2', D: 'b3', E: 'b4', F: 'c4', G: 'd4', H: 'e4'
+    });
+  });
+
+  it('should compose through concat', () => {
+    const e1 =   cold('--a--b-|');
+    const e2 =   cold(       '--x---y--|');
+    const expected =  '--a--b---x---y--|';
+
+    const result = MyCustomObservable.from(e1).concat(e2, rxTestScheduler);
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    expectObservable(result).toBe(expected);
+  });
+
+  it('should compose through merge', () => {
+    const e1 =   cold('-a--b-| ');
+    const e2 =   cold('--x--y-|');
+    const expected =  '-ax-by-|';
+
+    const result = MyCustomObservable.from(e1).merge(e2, rxTestScheduler);
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    expectObservable(result).toBe(expected);
+  });
+
+  it('should compose through race', () => {
+    const e1 =  cold('---a-----b-----c----|');
+    const e1subs =   '^                   !';
+    const e2 =  cold('------x-----y-----z----|');
+    const e2subs =   '^  !';
+    const expected = '---a-----b-----c----|';
+
+    const result = MyCustomObservable.from(e1).race(e2);
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    expectObservable(result).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
+  it('should compose through zip', () => {
+    const e1 =   cold('-a--b-----c-d-e-|');
+    const e2 =   cold('--1--2-3-4---|   ');
+    const expected = ('--A--B----C-D|   ');
+
+    const result = MyCustomObservable.from(e1).zip(e2, (a: any, b: any) => String(a) + String(b));
+
+    expect(result instanceof MyCustomObservable).to.be.true;
+
+    expectObservable(result).toBe(expected, {
+      A: 'a1', B: 'b2', C: 'c3', D: 'd4'
+    });
+  });
+
   it('should allow injecting behaviors into all subscribers in an operator ' +
   'chain when overridden', (done: MochaDone) => {
     // The custom Subscriber

spec/observables/concat-spec.ts

@@ -342,4 +342,24 @@ describe('Observable.concat', () => {
       expect(vals).to.equal(r.shift());
     }, null, done);
   });
-});
+
+  it('should use the scheduler even when one Observable is concat\'d', (done: MochaDone) => {
+    let e1Subscribed = false;
+    const e1 = Observable.defer(() => {
+      e1Subscribed = true;
+      return Observable.of('a');
+    });
+
+    Observable
+      .concat(e1, Rx.Scheduler.async)
+      .subscribe({
+        error: done,
+        complete: () => {
+          expect(e1Subscribed).to.be.true;
+          done();
+        }
+      });
+
+    expect(e1Subscribed).to.be.false;
+  });
+});

spec/observables/merge-spec.ts

@@ -231,4 +231,24 @@ describe('Observable.merge(...observables, Scheduler, number)', () => {
     const expected = '-d-a-e-b-f-c---x---y---z---|';
     expectObservable(Observable.merge(e1, e2, e3, 2, rxTestScheduler)).toBe(expected);
   });
+
+  it('should use the scheduler even when one Observable is merged', (done: MochaDone) => {
+    let e1Subscribed = false;
+    const e1 = Observable.defer(() => {
+      e1Subscribed = true;
+      return Observable.of('a');
+    });
+
+    Observable
+      .merge(e1, Rx.Scheduler.async)
+      .subscribe({
+        error: done,
+        complete: () => {
+          expect(e1Subscribed).to.be.true;
+          done();
+        }
+      });
+
+    expect(e1Subscribed).to.be.false;
+  });
 });

spec/operators/combineLatest-spec.ts

@@ -6,8 +6,8 @@ const Observable = Rx.Observable;
 /** @test {combineLatest} */
 describe('Observable.prototype.combineLatest', () => {
   asDiagram('combineLatest')('should combine events from two cold observables', () => {
-    const e1 =   hot('-a--b-----c-d-e-|');
-    const e2 =   hot('--1--2-3-4---|   ');
+    const e1 =   cold('-a--b-----c-d-e-|');
+    const e2 =   cold('--1--2-3-4---|   ');
     const expected = '--A-BC-D-EF-G-H-|';
 
     const result = e1.combineLatest(e2, (a: any, b: any) => String(a) + String(b));

spec/operators/finally-spec.ts

@@ -160,4 +160,4 @@ describe('Observable.prototype.finally', () => {
     rxTestScheduler.flush();
     expect(executed).to.be.true;
   });
-});
+});

spec/operators/publish-spec.ts

@@ -18,9 +18,12 @@ describe('Observable.prototype.publish', () => {
     published.connect();
   });
 
-  it('should return a ConnectableObservable', () => {
+  it('should return a ConnectableObservable-ish', () => {
     const source = Observable.of(1).publish();
-    expect(source instanceof Rx.ConnectableObservable).to.be.true;
+    expect(typeof (<any> source)._subscribe === 'function').to.be.true;
+    expect(typeof (<any> source).getSubject === 'function').to.be.true;
+    expect(typeof source.connect === 'function').to.be.true;
+    expect(typeof source.refCount === 'function').to.be.true;
   });
 
   it('should do nothing if connect is not called, despite subscriptions', () => {

spec/operators/publishBehavior-spec.ts

@@ -18,9 +18,12 @@ describe('Observable.prototype.publishBehavior', () => {
     published.connect();
   });
 
-  it('should return a ConnectableObservable', () => {
+  it('should return a ConnectableObservable-ish', () => {
     const source = Observable.of(1).publishBehavior(1);
-    expect(source instanceof Rx.ConnectableObservable).to.be.true;
+    expect(typeof (<any> source)._subscribe === 'function').to.be.true;
+    expect(typeof (<any> source).getSubject === 'function').to.be.true;
+    expect(typeof source.connect === 'function').to.be.true;
+    expect(typeof source.refCount === 'function').to.be.true;
   });
 
   it('should only emit default value if connect is not called, despite subscriptions', () => {
@@ -327,4 +330,4 @@ describe('Observable.prototype.publishBehavior', () => {
     expect(results).to.deep.equal([]);
     done();
   });
-});
+});

spec/operators/publishLast-spec.ts

@@ -18,10 +18,12 @@ describe('Observable.prototype.publishLast', () => {
     published.connect();
   });
 
-  it('should return a ConnectableObservable', () => {
+  it('should return a ConnectableObservable-ish', () => {
     const source = Observable.of(1).publishLast();
-
-    expect(source instanceof Rx.ConnectableObservable).to.be.true;
+    expect(typeof (<any> source)._subscribe === 'function').to.be.true;
+    expect(typeof (<any> source).getSubject === 'function').to.be.true;
+    expect(typeof source.connect === 'function').to.be.true;
+    expect(typeof source.refCount === 'function').to.be.true;
   });
 
   it('should do nothing if connect is not called, despite subscriptions', () => {
@@ -249,4 +251,4 @@ describe('Observable.prototype.publishLast', () => {
     expect(subscriptions).to.equal(1);
     done();
   });
-});
+});

spec/operators/publishReplay-spec.ts

@@ -18,9 +18,12 @@ describe('Observable.prototype.publishReplay', () => {
     published.connect();
   });
 
-  it('should return a ConnectableObservable', () => {
+  it('should return a ConnectableObservable-ish', () => {
     const source = Observable.of(1).publishReplay();
-    expect(source instanceof Rx.ConnectableObservable).to.be.true;
+    expect(typeof (<any> source)._subscribe === 'function').to.be.true;
+    expect(typeof (<any> source).getSubject === 'function').to.be.true;
+    expect(typeof source.connect === 'function').to.be.true;
+    expect(typeof source.refCount === 'function').to.be.true;
   });
 
   it('should do nothing if connect is not called, despite subscriptions', () => {

spec/operators/race-spec.ts

@@ -20,6 +20,20 @@ describe('Observable.prototype.race', () => {
     expectSubscriptions(e2.subscriptions).toBe(e2subs);
   });
 
+  it('should race cold and cold and accept an Array of Observable argument', () => {
+    const e1 =  cold('---a-----b-----c----|');
+    const e1subs =   '^                   !';
+    const e2 =  cold('------x-----y-----z----|');
+    const e2subs =   '^  !';
+    const expected = '---a-----b-----c----|';
+
+    const result = e1.race([e2]);
+
+    expectObservable(result).toBe(expected);
+    expectSubscriptions(e1.subscriptions).toBe(e1subs);
+    expectSubscriptions(e2.subscriptions).toBe(e2subs);
+  });
+
   it('should race hot and hot', () => {
     const e1 =   hot('---a-----b-----c----|');
     const e1subs =   '^                   !';

src/Rx.ts

@@ -150,7 +150,6 @@ export {Subscriber} from './Subscriber';
 export {AsyncSubject} from './AsyncSubject';
 export {ReplaySubject} from './ReplaySubject';
 export {BehaviorSubject} from './BehaviorSubject';
-export {MulticastObservable} from './observable/MulticastObservable';
 export {ConnectableObservable} from './observable/ConnectableObservable';
 export {Notification} from './Notification';
 export {EmptyError} from './util/EmptyError';

src/observable/ConnectableObservable.ts

@@ -51,6 +51,15 @@ export class ConnectableObservable<T> extends Observable<T> {
   }
 }
 
+export const connectableObservableDescriptor: PropertyDescriptorMap = {
+  operator: { value: null },
+  _refCount: { value: 0, writable: true },
+  _subscribe: { value: (<any> ConnectableObservable.prototype)._subscribe },
+  getSubject: { value: (<any> ConnectableObservable.prototype).getSubject },
+  connect: { value: (<any> ConnectableObservable.prototype).connect },
+  refCount: { value: (<any> ConnectableObservable.prototype).refCount }
+};
+
 class ConnectableSubscriber<T> extends SubjectSubscriber<T> {
   constructor(destination: Subject<T>,
               private connectable: ConnectableObservable<T>) {

src/operator/combineLatest.ts

@@ -62,7 +62,7 @@ export function combineLatest<T, R>(...observables: Array<ObservableInput<any> |
 
   observables.unshift(this);
 
-  return new ArrayObservable(observables).lift(new CombineLatestOperator(project));
+  return this.lift.call(new ArrayObservable(observables), new CombineLatestOperator(project));
 }
 
 /* tslint:disable:max-line-length */

src/operator/concat.ts

@@ -45,7 +45,7 @@ import { MergeAllOperator } from './mergeAll';
  * @owner Observable
  */
 export function concat<T, R>(...observables: Array<ObservableInput<any> | Scheduler>): Observable<R> {
-  return concatStatic<T, R>(this, ...observables);
+  return this.lift.call(concatStatic<T, R>(this, ...observables));
 }
 
 /* tslint:disable:max-line-length */
@@ -119,5 +119,9 @@ export function concatStatic<T, R>(...observables: Array<ObservableInput<any> |
     scheduler = args.pop();
   }
 
+  if (scheduler === null && observables.length === 1) {
+    return <Observable<R>>observables[0];
+  }
+
   return new ArrayObservable(observables, scheduler).lift(new MergeAllOperator<R>(1));
 }