Skip to content

Commit

Permalink
feat(share): use another observable to control resets (tests)
Browse files Browse the repository at this point in the history
  • Loading branch information
@backbone87
backbone87 committed May 9, 2021
1 parent 337bf4c commit 778a9e7
Showing 1 changed file with 200 additions and 23 deletions.
223 changes: 200 additions & 23 deletions spec/operators/share-spec.ts
View file
Expand Up @@ -5,6 +5,7 @@ import {
map,
mergeMap,
mergeMapTo,
onErrorResumeNext,
repeat,
retry,
share,
Expand All @@ -17,12 +18,25 @@ import {
} from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';
import { observableMatcher } from '../helpers/observableMatcher';
import sinon = require('sinon');
import { SinonSpy, spy } from 'sinon';

const syncNotify = of(1);
const asapNotify = scheduled(syncNotify, asapScheduler);
const syncError = throwError(() => new Error());

function spyOnUnhandledError(fn: (spy: SinonSpy) => void): void {
const prevOnUnhandledError = config.onUnhandledError;

try {
const onUnhandledError = spy();
config.onUnhandledError = onUnhandledError;

fn(onUnhandledError);
} finally {
config.onUnhandledError = prevOnUnhandledError;
}
}

/** @test {share} */
describe('share', () => {
let rxTest: TestScheduler;
Expand Down Expand Up @@ -511,12 +525,6 @@ describe('share', () => {
resetOnComplete: () => NEVER,
resetOnRefCountZero: () => NEVER,
},
{
title: 'share(config) using throwError() as sync reset notifier equivalents',
resetOnError: () => syncError,
resetOnComplete: () => syncError,
resetOnRefCountZero: () => syncError,
},
]) {
describe(title, () => {
it('should not reset on error if configured to do so', () => {
Expand Down Expand Up @@ -607,7 +615,7 @@ describe('share', () => {

describe('share(config)', () => {
it('should use the connector function provided', () => {
const connector = sinon.spy(() => new Subject());
const connector = spy(() => new Subject());

rxTest.run(({ hot, expectObservable }) => {
const source = hot(' ---v---v---v---E--v---v---v---C---v----v--------v----v---');
Expand Down Expand Up @@ -657,37 +665,206 @@ describe('share', () => {

it('should reset on refCount 0 only after reset notifier emitted', () => {
rxTest.run(({ hot, cold, expectObservable, expectSubscriptions }) => {
const source = hot(' ---1---2---3---4---5---|');
const expected = ' ---1---2---3---4---5---|';
const subscription = '^-----------------------';
const source = hot(' ---1---2---3---4---5---|');
const sourceSubs = [
' ^----------------!',
' ^----------------! ',
// break the line, please
' ------------------^----!',
' ------------------^----!',
];
const expected = ' ---1---2---3---4---5---|';
const subscription = ' ^-----------------------';
const firstPause = cold(' -| ');
const reset = cold(' --r ');
const secondPause = cold(' ---| ');
// reset: ' --r '

const sharedSource = source.pipe(share({ resetOnRefCountZero: () => cold('--a|') }), take(2));
const sharedSource = source.pipe(share({ resetOnRefCountZero: () => reset }), take(2));

const result = concat(sharedSource, cold('-|'), sharedSource, cold('---|'), sharedSource);
const result = concat(sharedSource, firstPause, sharedSource, secondPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
});
});

it('should not reset on refCount 0 after reset notifier errored?', () => {
rxTest.run(({ hot, cold, expectObservable, expectSubscriptions }) => {
const source = hot(' ---1---2---3---4---5---|');
const expected = ' ---1---2---3---4---5---|';
const subscription = '^-----------------------';
const sourceSubs = ' ^----------------------!';
it('should reset on error only after reset notifier emitted', () => {
rxTest.run(({ cold, expectObservable, expectSubscriptions }) => {
const source = cold(' ---1---2---# ');
// source: ' ---1---2---# '
const sourceSubs = [
' ^----------! ',
// break the line, please
' --------------^----------! ',
];
const expected = ' ---1---2---------1---2----# ';
const subscription = ' ^-------------------------- ';
const firstPause = cold(' -------| ');
const reset = cold(' --r ');
const secondPause = cold(' -----| ');
// reset: ' --r'

const sharedSource = source.pipe(share({ resetOnError: () => reset, resetOnRefCountZero: false }), take(2));

const result = concat(sharedSource, firstPause, sharedSource, secondPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
});
});

it('should reset on complete only after reset notifier emitted', () => {
rxTest.run(({ cold, expectObservable, expectSubscriptions }) => {
const source = cold(' ---1---2---| ');
// source: ' ---1---2---| '
const sourceSubs = [
' ^----------! ',
// break the line, please
' --------------^----------! ',
];
const expected = ' ---1---2---------1---2----| ';
const subscription = ' ^-------------------------- ';
const firstPause = cold(' -------| ');
const reset = cold(' --r ');
const secondPause = cold(' -----| ');
// reset: ' --r'

const sharedSource = source.pipe(share({ resetOnComplete: () => reset, resetOnRefCountZero: false }), take(2));

const result = concat(sharedSource, firstPause, sharedSource, secondPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
});
});

it('should not reset on refCount 0 if reset notifier errors before emitting any value', () => {
spyOnUnhandledError((onUnhandledError) => {
const error = new Error();

rxTest.run(({ hot, cold, expectObservable, expectSubscriptions }) => {
const source = hot(' ---1---2---3---4---(5 )---|');
const sourceSubs = ' ^------------------(- )---!';
const expected = ' ---1---2-------4---(5|) ';
const subscription = ' ^------------------(- ) ';
const firstPause = cold(' ------| ');
const reset = cold(' --# ', undefined, error);
// reset: ' (- )-# '

const sharedSource = source.pipe(share({ resetOnRefCountZero: () => reset }), take(2));

const result = concat(sharedSource, firstPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
});

expect(onUnhandledError).to.have.been.calledTwice;
expect(onUnhandledError.getCall(0)).to.have.been.calledWithExactly(error);
expect(onUnhandledError.getCall(1)).to.have.been.calledWithExactly(error);
});
});

it('should not reset on error if reset notifier errors before emitting any value', () => {
spyOnUnhandledError((onUnhandledError) => {
const error = new Error();

rxTest.run(({ cold, expectObservable, expectSubscriptions }) => {
const source = cold(' ---1---2---# ');
const sourceSubs = ' ^----------! ';
const expected = ' ---1---2------#';
const subscription = ' ^--------------';
const firstPause = cold(' -------|');
const reset = cold(' --# ', undefined, error);

const sharedSource = source.pipe(share({ resetOnError: () => reset, resetOnRefCountZero: false }), take(2));

const result = concat(sharedSource, firstPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
});

expect(onUnhandledError).to.have.been.calledOnce;
expect(onUnhandledError.getCall(0)).to.have.been.calledWithExactly(error);
});
});

it('should not reset on complete if reset notifier errors before emitting any value', () => {
spyOnUnhandledError((onUnhandledError) => {
const error = new Error();

rxTest.run(({ cold, expectObservable, expectSubscriptions }) => {
const source = cold(' ---1---2---| ');
const sourceSubs = ' ^----------! ';
const expected = ' ---1---2------|';
const subscription = ' ^--------------';
const firstPause = cold(' -------|');
const reset = cold(' --# ', undefined, error);

const sharedSource = source.pipe(share({ resetOnComplete: () => reset, resetOnRefCountZero: false }), take(2));

const result = concat(sharedSource, firstPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
});

expect(onUnhandledError).to.have.been.calledOnce;
expect(onUnhandledError.getCall(0)).to.have.been.calledWithExactly(error);
});
});

it('should not call "resetOnRefCountZero" on error', () => {
rxTest.run(({ cold, expectObservable, expectSubscriptions }) => {
const resetOnRefCountZero = spy(() => EMPTY);

const source = cold(' ---1---(2#) ');
// source: ' ---1---(2#) '
const sourceSubs = [
' ^------(! ) ',
// break the line, please
' -------(- )---^------(! ) ',
];
const expected = ' ---1---(2 )------1---(2#) ';
const subscription = ' ^------(- )----------(- ) ';
const firstPause = cold(' (- )---| ');
const reset = cold(' (- )-r ');
// reset: ' (- )-r'

const sharedSource = source.pipe(share({ resetOnError: () => reset, resetOnRefCountZero }));

const result = concat(sharedSource.pipe(onErrorResumeNext(firstPause)), sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
expect(resetOnRefCountZero).to.not.have.been.called;
});
});

it('should not call "resetOnRefCountZero" on complete', () => {
rxTest.run(({ cold, expectObservable, expectSubscriptions }) => {
const resetOnRefCountZero = spy(() => EMPTY);

const source = cold(' ---1---(2|) ');
// source: ' ---1---(2|) '
const sourceSubs = [
' ^------(! ) ',
// break the line, please
' -------(- )---^------(! ) ',
];
const expected = ' ---1---(2 )------1---(2|) ';
const subscription = ' ^------(- )----------(- ) ';
const firstPause = cold(' (- )---| ');
const reset = cold(' (- )-r ');
// reset: ' (- )-r'

const sharedSource = source.pipe(share({ resetOnRefCountZero: () => cold('--#') }), take(2));
const sharedSource = source.pipe(share({ resetOnComplete: () => reset, resetOnRefCountZero }));

const result = concat(sharedSource, cold('-|'), sharedSource, cold('---|'), sharedSource);
const result = concat(sharedSource, firstPause, sharedSource);

expectObservable(result, subscription).toBe(expected);
expectSubscriptions(source.subscriptions).toBe(sourceSubs);
expect(resetOnRefCountZero).to.not.have.been.called;
});
});
});
Expand Down

0 comments on commit 778a9e7

Please sign in to comment.