RxJS Echo Operator - re-emits (echoes) the last emitted value

echo.test.ts

import { NEVER, of } from 'rxjs';
import { filter, switchMap } from 'rxjs/operators';
import { TestScheduler } from 'rxjs/testing';
import { echo, echoGroup } from './echo';

const testScheduler = () =>
  new TestScheduler((actual, expected) => {
    expect(actual).toEqual(expected);
  });

it('echoes the value via timer', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('-a 120000ms |');
    // prettier-ignore
    const e1subs = [
      '              ^- 120000ms ! ',
      '              -^- 119999ms !'
    ];

    const expected = '-a 59999ms a 59999ms a|';

    expectObservable(e1.pipe(echo())).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('configures echoes via timer', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('  -a------------| ');
    // prettier-ignore
    const e1subs = [
      '                ^-------------!',
      '                -^------------!'
    ];
    const expected = '-a----a----a--| ';

    expectObservable(
      e1.pipe(
        echo({
          timerTrigger: 5,
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('echoes multiple values via timer and switches to latest', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('  -a--b---------c-|');
    // prettier-ignore
    const e1subs = [
      '                ^---------------!',
      '                -^--!',
      '                ----^---------!',
      '                --------------^-!',
    ];
    const expected = '-a--b----b----c-|';

    expectObservable(
      e1.pipe(
        echo({
          timerTrigger: 5,
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('disables echoes of values via timer', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('  -a---| ');
    // prettier-ignore
    const e1subs = [
      '                ^----!',
      '                -^---!'
    ];
    const expected = '-a---| ';

    expectObservable(
      e1.pipe(
        echo({
          timerTrigger: false,
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('restart echoes on event', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('   -a-----| ');
    const _evts = hot('----e--| ').subscribe(() => {
      window.dispatchEvent(new Event('focus'));
    });
    // prettier-ignore
    const e1subs = [
      '                 ^------!',
      '                 -^-----!',
    ];
    const expected = ' -a--a--| ';

    expectObservable(
      e1.pipe(
        echo({
          timerTrigger: 5,
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('echoes the value via focus', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('   -a----------| ');
    const _evts = hot('----e--e-e--| ').subscribe(() => {
      window.dispatchEvent(new Event('focus'));
    });
    // prettier-ignore
    const e1subs = [
      '                 ^-----------!',
      '                 -^----------!'];
    const expected = ' -a--a--a-a--| ';

    expectObservable(e1.pipe(echo())).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('disables echoes the value via focus', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('   -a-------| ');
    const _evts = hot('----e----| ').subscribe(() => {
      window.dispatchEvent(new Event('focus'));
    });
    // prettier-ignore
    const e1subs = [
          '             ^--------!',
          '             -^-------!'
    ];
    const expected = ' -a-------| ';

    expectObservable(
      e1.pipe(
        echo({
          focusTrigger: false,
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('echoes the value via online', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('   -a----------| ');
    const _evts = hot('----e--e-e--| ').subscribe(() => {
      window.dispatchEvent(new Event('online'));
    });
    // prettier-ignore
    const e1subs = [
      '                 ^-----------!',
      '                 -^----------!'];
    const expected = ' -a--a--a-a--| ';

    expectObservable(e1.pipe(echo())).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('disables echoes the value via online', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('   -a-------| ');
    const _evts = hot('----e----| ').subscribe(() => {
      window.dispatchEvent(new Event('online'));
    });
    // prettier-ignore
    const e1subs = [
      '                 ^--------!',
      '                 -^-------!'
    ];
    const expected = ' -a-------| ';

    expectObservable(
      e1.pipe(
        echo({
          onlineTrigger: false,
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('echoes can be triggered by custom events', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('  -a----------| ');
    const evts = hot('----e----e--| ');
    // prettier-ignore
    const e1subs = [
      '                ^-----------!',
      '                -^----------!',
    ];
    const expected = '-a--a----a--| ';

    expectObservable(
      e1.pipe(
        echo({
          triggers: () => [evts],
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('group configures echoes of values via timer', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('  -a------------| ');
    // prettier-ignore
    const e1subs = [
      '                ^-------------!',
      '                -^------------!'
    ];
    const expected = '-a----a----a--| ';

    expectObservable(
      e1.pipe(
        echoGroup({
          grouper: (v) => v,
          runner: switchMap((v) => of(v)),
          stopper: (value) => e1.pipe(filter((value2) => value !== value2)),
          triggers: {
            timerTrigger: 5,
          },
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('group creates multiple groups', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot(' -a------b-----| ');
    const e1subs = [' ^-------------!'];
    const expected = '-a----a-b--a-b|';

    expectObservable(
      e1.pipe(
        echoGroup({
          grouper: (v) => v,
          runner: switchMap((v) => of(v)),
          stopper: (_value) => NEVER,
          triggers: {
            timerTrigger: 5,
          },
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

it('group creates multiple groups and completes previous created groups', () => {
  testScheduler().run(({ hot, expectObservable, expectSubscriptions }) => {
    const e1 = hot('  -a------b-----| ');
    // prettier-ignore
    const e1subs = [
      '                ^-------------!',
      '                -^------!      ',
      '                --------^-----!',
    ];
    const expected = '-a----a-b----b |';

    expectObservable(
      e1.pipe(
        echoGroup({
          grouper: (v) => v,
          runner: switchMap((v) => of(v)),
          stopper: (value) => e1.pipe(filter((value2) => value !== value2)),
          triggers: {
            timerTrigger: 5,
          },
        })
      )
    ).toBe(expected);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });
});

echo.ts

import {
  fromEvent,
  merge,
  MonoTypeOperatorFunction,
  NEVER,
  Observable,
  ObservableInput,
  ObservedValueOf,
  OperatorFunction,
  timer,
} from 'rxjs';
import {
  exhaustMap,
  filter,
  groupBy,
  mapTo,
  materialize,
  mergeMap,
  startWith,
  switchMap,
  take,
  takeUntil,
} from 'rxjs/operators';

export function echo<Value>({
  timerTrigger = 60_000,
  focusTrigger = true,
  onlineTrigger = true,
  triggers = () => [],
}: TriggerConfig<Value> = {}): MonoTypeOperatorFunction<Value> {
  return (source) => {
    const triggers$ = [
      focusTrigger ? fromEvent(window, 'focus') : NEVER,
      onlineTrigger ? fromEvent(window, 'online') : NEVER,
    ];
    return source.pipe(
      switchMap((value) => {
        return merge(...triggers$, ...triggers(value)).pipe(
          startWith(value),
          switchMap(() => timer(0, timerTrigger === false ? Infinity : timerTrigger)),
          mapTo(value),
          takeUntil(
            source.pipe(
              materialize(),
              filter(({ kind }) => kind !== 'N'),
            ),
          ),
        );
      }),
    );
  };
}

export function echoGroup<Value, Result extends ObservableInput<any>>({
  grouper,
  runner,
  stopper,
  triggers,
}: {
  grouper: (value: Value) => unknown;
  runner: OperatorFunction<Value, ObservedValueOf<Result>>;
  stopper: (value: Value) => Observable<unknown>;
  triggers?: TriggerConfig<Value>;
}): OperatorFunction<Value, ObservedValueOf<Result>> {
  return (source) => {
    return source.pipe(
      groupBy(
        grouper,
        (value) => value,
        (group$) =>
          group$.pipe(
            exhaustMap((value) => stopper(value)),
            take(1),
          ),
      ),
      mergeMap((group$) => group$.pipe(echo(triggers), runner)),
    );
  };
}

export interface TriggerConfig<Value> {
  focusTrigger?: boolean;
  onlineTrigger?: boolean;
  timerTrigger?: false | number;
  triggers?: (value: Value) => Observable<Value>[];
}