fix(multicast): Fixes multicast with selector to create a new source …

…connection per subscriber.

Multicast with a selector function should create a new ConnectableObservable for each subscriber to
the MulticastObservable. This ensures each subscriber creates a new connection to the source
Observable, and don't share subscription side-effects.

spec/operators/multicast-spec.ts

@@ -51,7 +51,9 @@ describe('Observable.prototype.multicast', () => {
 
  it('should accept selectors to factory functions', () => {
  const source = hot('-1-2-3----4-|');
- const sourceSubs = ['^ !'];
+ const sourceSubs = ['^ !',
+ ' ^ !',
+ ' ^ !'];
  const multicasted = source.multicast(() => new Subject(),
  x => x.zip(x, (a, b) => (parseInt(a) + parseInt(b)).toString()));
  const subscriber1 = hot('a| ').mergeMapTo(multicasted);

spec/operators/publish-spec.ts

@@ -54,7 +54,9 @@ describe('Observable.prototype.publish', () => {
 
  it('should accept selectors', () => {
  const source = hot('-1-2-3----4-|');
- const sourceSubs = ['^ !'];
+ const sourceSubs = ['^ !',
+ ' ^ !',
+ ' ^ !'];
  const published = source.publish(x => x.zip(x, (a, b) => (parseInt(a) + parseInt(b)).toString()));
  const subscriber1 = hot('a| ').mergeMapTo(published);
  const expected1 = '-2-4-6----8-|';
@@ -321,4 +323,4 @@ describe('Observable.prototype.publish', () => {
  expect(subscriptions).to.equal(1);
  done();
  });
-});
+});

spec/support/debug.opts

@@ -0,0 +1,14 @@
+--require source-map-support/register
+--require spec/support/mocha-setup-node.js
+--require spec-js/helpers/test-helper.js
+--require spec-js/helpers/ajax-helper.js
+--ui spec-js/helpers/testScheduler-ui.js
+
+--reporter dot
+--bail
+--full-trace
+--check-leaks
+--globals WebSocket,FormData
+
+--recursive
+--timeout 100000

src/observable/MulticastObservable.ts

@@ -1,20 +1,21 @@
+import {Subject} from '../Subject';
 import {Observable} from '../Observable';
 import {Subscriber} from '../Subscriber';
 import {Subscription} from '../Subscription';
 import {ConnectableObservable} from '../observable/ConnectableObservable';
 
 export class MulticastObservable<T> extends Observable<T> {
  constructor(protected source: Observable<T>,
- private connectable: ConnectableObservable<T>,
+ private subjectFactory: () => Subject<T>,
  private selector: (source: Observable<T>) => Observable<T>) {
  super();
  }
 
  protected _subscribe(subscriber: Subscriber<T>): Subscription {
- const {selector, connectable} = this;
-
+ const { selector, source } = this;
+ const connectable = new ConnectableObservable(source, this.subjectFactory);
  const subscription = selector(connectable).subscribe(subscriber);
  subscription.add(connectable.connect());
  return subscription;
  }
-}
+}

src/operator/multicast.ts

@@ -33,8 +33,9 @@ export function multicast<T>(subjectOrSubjectFactory: Subject<T> | (() => Subjec
  };
  }
 
- const connectable = new ConnectableObservable(this, subjectFactory);
- return selector ? new MulticastObservable(this, connectable, selector) : connectable;
+ return !selector ?
+ new ConnectableObservable(this, subjectFactory) :
+ new MulticastObservable(this, subjectFactory, selector);
 }
 
 export type factoryOrValue<T> = T | (() => T);