fix(publish,publishReplay) resolve sharing Subject

change publish operator to use factory
change publishReplay operator to not share ReplaySubject
fixes issue ReactiveX#5411

spec/operators/publish-spec.ts

@@ -1,7 +1,7 @@
 import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
 import { publish, zip, mergeMapTo, mergeMap, tap, refCount, retry, repeat } from 'rxjs/operators';
-import { ConnectableObservable, of, Subscription, Observable } from 'rxjs';
+import { ConnectableObservable, of, Subscription, Observable, pipe } from 'rxjs';
 
 /** @test {publish} */
 describe('publish operator', () => {
@@ -337,4 +337,28 @@ describe('publish operator', () => {
  expect(subscriptions).to.equal(1);
  done();
  });
+
+ it('should subscribe to its own source when using a shared pipeline', () => {
+ const source1 = cold('-1-2-3-4-5-|');
+ const source1Subs = '^ !';
+ const source2 = cold('-6-7-8-9-0-|');
+ const source2Subs = '^ !'; 
+
+ const sharedPipeLine = pipe(
+ publish()
+ );
+
+ const published1 = source1.pipe(sharedPipeLine) as ConnectableObservable<any>;
+ const published2 = source2.pipe(sharedPipeLine) as ConnectableObservable<any>;
+ const expected1 = '-1-2-3-4-5-|';
+ const expected2 = '-6-7-8-9-0-|';
+
+ expectObservable(published1).toBe(expected1);
+ expectSubscriptions(source1.subscriptions).toBe(source1Subs);
+ expectObservable(published2).toBe(expected2);
+ expectSubscriptions(source2.subscriptions).toBe(source2Subs);
+
+ published1.connect();
+ published2.connect();
+ });
 });

spec/operators/publishReplay-spec.ts

@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { hot, cold, expectObservable, expectSubscriptions } from '../helpers/marble-testing';
-import { throwError, ConnectableObservable, EMPTY, NEVER, of, Observable, Subscription } from 'rxjs';
+import { throwError, ConnectableObservable, EMPTY, NEVER, of, Observable, Subscription, pipe } from 'rxjs';
 import { publishReplay, mergeMapTo, tap, mergeMap, refCount, retry, repeat, map } from 'rxjs/operators';
 
 /** @test {publishReplay} */
@@ -488,4 +488,28 @@ describe('publishReplay operator', () => {
  expectObservable(published).toBe(expected, undefined, error);
  expectSubscriptions(source.subscriptions).toBe(sourceSubs);
  });
+
+ it('should subscribe to its own source when using a shared pipeline', () => {
+ const source1 = cold('-1-2-3-4-5-|');
+ const source2 = cold('-6-7-8-9-0-|');
+ const expected1 = '-1-2-3-4-5-|';
+ const expected2 = '-6-7-8-9-0-|';
+ const source1Subs = '^ !';
+ const source2Subs = '^ !';
+
+ const sharedPipeLine = pipe(
+ publishReplay(1)
+ );
+
+ const published1 = source1.pipe(sharedPipeLine) as ConnectableObservable<any>;
+ const published2 = source2.pipe(sharedPipeLine) as ConnectableObservable<any>;
+
+ expectObservable(published1).toBe(expected1);
+ expectSubscriptions(source1.subscriptions).toBe(source1Subs);
+ expectObservable(published2).toBe(expected2);
+ expectSubscriptions(source2.subscriptions).toBe(source2Subs);
+
+ published1.connect();
+ published2.connect();
+ });
 });

src/internal/operators/publish.ts

@@ -59,6 +59,6 @@ export function publish<T>(selector: MonoTypeOperatorFunction<T>): MonoTypeOpera
  */
 export function publish<T, R>(selector?: OperatorFunction<T, R>): MonoTypeOperatorFunction<T> | OperatorFunction<T, R> {
  return selector ?
- multicast(() => new Subject<T>(), selector) :
- multicast(new Subject<T>());
+ (source: Observable<T>) => multicast(() => new Subject<T>(), selector)(source) :
+ (source: Observable<T>) => multicast(new Subject<T>())(source);
 }

src/internal/operators/publishReplay.ts

@@ -19,7 +19,6 @@ export function publishReplay<T, R>(bufferSize?: number,
  }
 
  const selector = typeof selectorOrScheduler === 'function' ? selectorOrScheduler : undefined;
- const subject = new ReplaySubject<T>(bufferSize, windowTime, scheduler);
 
- return (source: Observable<T>) => multicast(() => subject, selector!)(source) as ConnectableObservable<R>;
+ return (source: Observable<T>) => multicast(new ReplaySubject<T>(bufferSize, windowTime, scheduler), selector!)(source) as ConnectableObservable<R>;
 }