-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
14 changed files
with
546 additions
and
161 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
148 changes: 148 additions & 0 deletions
148
spec/deprecation-equivalents/multicasting-deprecations-spec.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,148 @@ | ||
| /** @prettier */ | ||
| import { Observable, ConnectableObservable, connectable, of, AsyncSubject, BehaviorSubject, ReplaySubject, Subject, merge } from 'rxjs'; | ||
| import { connect, share, multicast, publish, publishReplay, publishBehavior, publishLast, refCount, repeat, retry } from 'rxjs/operators'; | ||
| import { TestScheduler } from 'rxjs/testing'; | ||
| import { observableMatcher } from '../helpers/observableMatcher'; | ||
|
|
||
| describe('multicasting equivalent tests', () => { | ||
| let rxTest: TestScheduler; | ||
|
|
||
| beforeEach(() => { | ||
| rxTest = new TestScheduler(observableMatcher); | ||
| }); | ||
|
|
||
| testEquivalents( | ||
| 'multicast(() => new Subject()), refCount() and share()', | ||
| (source) => | ||
| source.pipe( | ||
| multicast(() => new Subject<string>()), | ||
| refCount() | ||
| ), | ||
| (source) => source.pipe(share()) | ||
| ); | ||
|
|
||
| testEquivalents( | ||
| 'multicast(new Subject()), refCount() and share({ resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })', | ||
| (source) => source.pipe(multicast(new Subject()), refCount()), | ||
| (source) => source.pipe(share({ resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })) | ||
| ); | ||
|
|
||
| testEquivalents( | ||
| 'publish(), refCount() and share({ resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })', | ||
| (source) => source.pipe(publish(), refCount()), | ||
| (source) => source.pipe(share({ resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })) | ||
| ); | ||
|
|
||
| testEquivalents( | ||
| 'publishLast(), refCount() and share({ connector: () => new AsyncSubject(), resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })', | ||
| (source) => source.pipe(publishLast(), refCount()), | ||
| (source) => | ||
| source.pipe(share({ connector: () => new AsyncSubject(), resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })) | ||
| ); | ||
|
|
||
| testEquivalents( | ||
| 'publishBehavior("X"), refCount() and share({ connector: () => new BehaviorSubject("X"), resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })', | ||
| (source) => source.pipe(publishBehavior('X'), refCount()), | ||
| (source) => | ||
| source.pipe( | ||
| share({ connector: () => new BehaviorSubject('X'), resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false }) | ||
| ) | ||
| ); | ||
|
|
||
| testEquivalents( | ||
| 'publishReplay(3, 10), refCount() and share({ connector: () => new ReplaySubject(3, 10), resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false })', | ||
| (source) => source.pipe(publishReplay(3, 10), refCount()), | ||
| (source) => | ||
| source.pipe( | ||
| share({ connector: () => new ReplaySubject(3, 10), resetOnError: false, resetOnComplete: false, resetOnUnsubscribe: false }) | ||
| ) | ||
| ); | ||
|
|
||
| const fn = (source: Observable<any>) => merge(source, source); | ||
|
|
||
| testEquivalents( | ||
| 'publish(fn) and connect({ setup: fn })', | ||
| (source) => source.pipe(publish(fn)), | ||
| (source) => | ||
| source.pipe( | ||
| connect({ | ||
| setup: fn, | ||
| }) | ||
| ) | ||
| ); | ||
|
|
||
| testEquivalents( | ||
| 'publishReplay(3, 10, fn) and `subject = new ReplaySubject(3, 10), connect({ connector: () => subject , setup: fn })`', | ||
| (source) => source.pipe(publishReplay(3, 10, fn)), | ||
| (source) => { | ||
| const subject = new ReplaySubject(3, 10); | ||
| return source.pipe(connect({ connector: () => subject, setup: fn })); | ||
| } | ||
| ); | ||
|
|
||
| /** | ||
| * Used to test a variety of scenarios with multicast operators that should be equivalent. | ||
| * @param name The name to add to the test output | ||
| * @param oldExpression The old expression we're saying matches the updated expression | ||
| * @param updatedExpression The updated expression we're telling people to use instead. | ||
| */ | ||
| function testEquivalents( | ||
| name: string, | ||
| oldExpression: (source: Observable<string>) => Observable<string>, | ||
| updatedExpression: (source: Observable<string>) => Observable<string> | ||
| ) { | ||
| it(`should be equivalent for ${name} for async sources`, () => { | ||
| rxTest.run(({ cold, expectObservable }) => { | ||
| const source = cold('----a---b---c----d---e----|'); | ||
| const old = oldExpression(source); | ||
| const updated = updatedExpression(source); | ||
| expectObservable(updated).toEqual(old); | ||
| }); | ||
| }); | ||
|
|
||
| it(`should be equivalent for ${name} for async sources that repeat`, () => { | ||
| rxTest.run(({ cold, expectObservable }) => { | ||
| const source = cold('----a---b---c----d---e----|'); | ||
| const old = oldExpression(source).pipe(repeat(3)); | ||
| const updated = updatedExpression(source).pipe(repeat(3)); | ||
| expectObservable(updated).toEqual(old); | ||
| }); | ||
| }); | ||
|
|
||
| it(`should be equivalent for ${name} for async sources that retry`, () => { | ||
| rxTest.run(({ cold, expectObservable }) => { | ||
| const source = cold('----a---b---c----d---e----#'); | ||
| const old = oldExpression(source).pipe(retry(3)); | ||
| const updated = updatedExpression(source).pipe(retry(3)); | ||
| expectObservable(updated).toEqual(old); | ||
| }); | ||
| }); | ||
|
|
||
| it(`should be equivalent for ${name} for async sources`, () => { | ||
| rxTest.run(({ expectObservable }) => { | ||
| const source = of('a', 'b', 'c'); | ||
| const old = oldExpression(source); | ||
| const updated = updatedExpression(source); | ||
| expectObservable(updated).toEqual(old); | ||
| }); | ||
| }); | ||
|
|
||
| it(`should be equivalent for ${name} for async sources that repeat`, () => { | ||
| rxTest.run(({ expectObservable }) => { | ||
| const source = of('a', 'b', 'c'); | ||
| const old = oldExpression(source).pipe(repeat(3)); | ||
| const updated = updatedExpression(source).pipe(repeat(3)); | ||
| expectObservable(updated).toEqual(old); | ||
| }); | ||
| }); | ||
|
|
||
| it(`should be equivalent for ${name} for async sources that retry`, () => { | ||
| rxTest.run(({ expectObservable }) => { | ||
| const source = of('a', 'b', 'c'); | ||
| const old = oldExpression(source).pipe(retry(3)); | ||
| const updated = updatedExpression(source).pipe(retry(3)); | ||
| expectObservable(updated).toEqual(old); | ||
| }); | ||
| }); | ||
| } | ||
| }); |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,44 +1,110 @@ | ||
| /** @prettier */ | ||
| import { lift } from '../util/lift'; | ||
| import { Subscriber } from '../Subscriber'; | ||
| import { OperatorFunction, ObservableInput } from '../types'; | ||
| import { Observable } from '../Observable'; | ||
| import { Subject } from '../Subject'; | ||
| import { from } from '../observable/from'; | ||
| import { Subscription } from '../Subscription'; | ||
| import { operate } from '../util/lift'; | ||
|
|
||
| /** | ||
| * The default connector function used for `connect`. | ||
| * A factory function that will create a {@link Subject}. | ||
| */ | ||
| function defaultConnector<T>() { | ||
| return new Subject<T>(); | ||
| } | ||
|
|
||
| /** | ||
| * Creates an observable by multicasting the source within a function that | ||
| * allows the developer to define the usage of the multicast prior to connection. | ||
| * | ||
| * This is particularly useful if the observable source you wish to multicast could | ||
| * be synchronous or asynchronous. This sets it apart from {@link share}, which, in the | ||
| * case of totally synchronous sources will fail to share a single subscription with | ||
| * multiple consumers, as by the time the subscription to the result of {@link share} | ||
| * has returned, if the source is synchronous its internal reference count will jump from | ||
| * 0 to 1 back to 0 and reset. | ||
| * | ||
| * To use `connect`, you provide a `setup` function via configuration that will give you | ||
| * a multicast observable that is not yet connected. You then use that multicast observable | ||
| * to create a resulting observable that, when subscribed, will set up your multicast. This is | ||
| * generally, but not always, accomplished with {@link merge}. | ||
| * | ||
| * Note that using a {@link takeUntil} inside of `connect`'s `setup` _might_ mean you were looking | ||
| * to use the {@link takeWhile} operator instead. | ||
| * | ||
| * When you subscribe to the result of `connect`, the `setup` function will be called. After | ||
| * the `setup` function returns, the observable it returns will be subscribed to, _then_ the | ||
| * multicast will be connected to the source. | ||
| * | ||
| * ### Example | ||
| * | ||
| * Sharing a totally synchronous observable | ||
| * | ||
| * ```ts | ||
| * import { defer, of } from 'rxjs'; | ||
| * import { tap, connect } from 'rxjs/operators'; | ||
| * | ||
| * const source$ = defer(() => { | ||
| * console.log('subscription started'); | ||
| * return of(1, 2, 3, 4, 5).pipe( | ||
| * tap(n => console.log(`source emitted ${n}`)) | ||
| * ); | ||
| * }); | ||
| * | ||
| * source$.pipe( | ||
| * connect({ | ||
| * // Notice in here we're merging three subscriptions to `shared$`. | ||
| * setup: (shared$) => merge( | ||
| * shared$.pipe(map(n => `all ${n}`)), | ||
| * shared$.pipe(filter(n => n % 2 === 0), map(n => `even ${n}`)), | ||
| * shared$.pipe(filter(n => n % 2 === 1), map(n => `odd ${n}`)), | ||
| * ) | ||
| * }) | ||
| * ) | ||
| * .subscribe(console.log); | ||
| * | ||
| * // Expected output: (notice only one subscription) | ||
| * "subscription started" | ||
| * "source emitted 1" | ||
| * "all 1" | ||
| * "odd 1" | ||
| * "source emitted 2" | ||
| * "all 2" | ||
| * "even 2" | ||
| * "source emitted 3" | ||
| * "all 3" | ||
| * "odd 3" | ||
| * "source emitted 4" | ||
| * "all 4" | ||
| * "even 4" | ||
| * "source emitted 5" | ||
| * "all 5" | ||
| * "odd 5" | ||
| * ``` | ||
| * | ||
| * @param param0 The configuration object for `connect`. | ||
| */ | ||
| export function connect<T, R>({ | ||
| connector = defaultConnector, | ||
| setup, | ||
| }: { | ||
| /** | ||
| * A factory function used to create the Subject through which the source | ||
| * is multicast. By default this creates a {@link Subject}. | ||
| */ | ||
| connector?: () => Subject<T>; | ||
| /** | ||
| * A function used to set up the multicast. Gives you a multicast observable | ||
| * that is not yet connected. With that, you're expected to create and return | ||
| * and Observable, that when subscribed to, will utilize the multicast observable. | ||
| * After this function is executed -- and its return value subscribed to -- the | ||
| * the operator will subscribe to the source, and the connection will be made. | ||
| */ | ||
| setup: (shared: Observable<T>) => ObservableInput<R>; | ||
| }): OperatorFunction<T, R> { | ||
| return (source: Observable<T>) => { | ||
| return lift(source, function (this: Subscriber<R>, source: Observable<T>) { | ||
| const subscriber = this; | ||
| let subject: Subject<T>; | ||
| try { | ||
| subject = connector(); | ||
| } catch (err) { | ||
| subscriber.error(err); | ||
| return; | ||
| } | ||
|
|
||
| let result: Observable<R>; | ||
| try { | ||
| result = from(setup(subject.asObservable())); | ||
| } catch (err) { | ||
| subscriber.error(err); | ||
| return; | ||
| } | ||
|
|
||
| const subscription = result.subscribe(subscriber); | ||
| subscription.add(source.subscribe(subject)); | ||
| return subscription; | ||
| }); | ||
| }; | ||
| return operate((source, subscriber) => { | ||
| const subject = connector(); | ||
| from(setup(subject.asObservable())).subscribe(subscriber); | ||
| subscriber.add(source.subscribe(subject)); | ||
| }); | ||
| } |
Oops, something went wrong.