src/app/util/rxjs/rxjs.util.spec.ts

/********************************************************************************
 * Copyright (c) 2020 Contributors to the Eclipse Foundation
 *
 * See the NOTICE file(s) distributed with this work for additional
 * information regarding copyright ownership.
 *
 * This program and the accompanying materials are made available under the
 * terms of the Eclipse Public License 2.0 which is available at
 * http://www.eclipse.org/legal/epl-2.0
 *
 * SPDX-License-Identifier: EPL-2.0
 ********************************************************************************/

import {NgZone} from "@angular/core";
import {fakeAsync, tick} from "@angular/core/testing";
import {createAction} from "@ngrx/store";
import {concat, of, throwError, timer} from "rxjs";
import {map, tap, toArray} from "rxjs/operators";
import {
    catchErrorTo,
    emitOnComplete,
    endWithObservable,
    ignoreError,
    retryAfter,
    runInZone,
    runOutsideZone,
    throwAfterActionType
} from "./rxjs.util";

describe("retryAfter", () => {

    const timeInMs = 100;

    it("should not have effects when no error is thrown", fakeAsync(() => {
        const random = Math.random();
        const success$ = of(random);

        let gotResult = false;
        let hasFinished = false;

        const subscription = success$.pipe(
            retryAfter(timeInMs)
        ).subscribe({
            next: (result) => gotResult = result === random,
            complete: () => hasFinished = true
        });

        tick();

        expect(gotResult).toBeTrue();
        expect(hasFinished).toBeTrue();

        subscription.unsubscribe();
    }));

    it("should only retry after a specific time", fakeAsync(() => {
        const error = new Error("" + Math.random());

        let counter = 0;

        const throw$ = of(0).pipe(
            map(() => {
                counter++;
                throw error;
            }),
            retryAfter(timeInMs)
        );

        const subscription = throw$.subscribe();

        expect(counter).toBe(1);

        for (let i = 0; i < 100; i++) {
            tick(timeInMs - 1);
            expect(counter).toBe(i + 1);
            tick(1);
            expect(counter).toBe(i + 2);
        }

        subscription.unsubscribe();
    }));

    it("should only retry a specific amount of times if specified", fakeAsync(() => {
        const retryCount = 3;

        let lastError: any;
        let counter = 0;
        let hasFinished = false;
        let hasError = false;

        const throw$ = of(0).pipe(
            map(() => {
                counter++;
                lastError = new Error("" + Math.random());
                throw lastError;
            }),
            retryAfter(timeInMs, 2)
        );

        const subscription = throw$.subscribe({
            error: (err) => hasError = lastError === err,
            complete: () => hasFinished = true
        });

        tick(retryCount * timeInMs - 1);

        expect(hasFinished).toBeFalse();
        expect(hasError).toBeFalse();

        tick(1);

        expect(hasFinished).toBeFalse();
        expect(hasError).toBeTrue();
        expect(counter - 1).toBe(retryCount);

        subscription.unsubscribe();
    }));

});

describe("ignoreError", () => {

    it("should complete an throwing observable", async () => {
        const observable = throwError(new Error("Test Error")).pipe(ignoreError());
        await expectAsync(observable.toPromise()).not.toBeRejected();
    });

});

describe("catchErrorTo", () => {

    it("should catch errors and emit the given value", async () => {
        const observable = throwError(new Error("Test Error")).pipe(catchErrorTo(19));
        await expectAsync(observable.toPromise()).toBeResolvedTo(19);
    });

});

describe("throwAfterActionType", () => {

    const testAction = createAction("Test Action");
    const throwAction = createAction("Throw Action");

    it("should throw an error after emitting a given action type", async () => {
        const results: any[] = [];
        const observable = of(testAction(), throwAction()).pipe(throwAfterActionType(throwAction), tap((_) => results.push(_)));
        await expectAsync(observable.toPromise()).toBeRejectedWith(throwAction());
        expect(results).toEqual([testAction(), throwAction()]);
    });

});


describe("endWithObservable", () => {

    it("should end an observable with another observable", async () => {
        const observable$ = of(0).pipe(
            endWithObservable(() => of(1)),
            toArray()
        );
        const result = await observable$.toPromise();
        expect(result).toEqual([0, 1]);
    });

});

describe("emitOnComplete", () => {

    it("should emit all values on completion", fakeAsync(() => {
        const observable$ = concat(
            of("start"),
            timer(1000)
        ).pipe(emitOnComplete());

        const result: any[] = [];
        const subscription = observable$.subscribe((_) => result.push(_));

        tick(999);
        expect(subscription.closed).toBeFalse();
        expect(result).toEqual([]);
        tick(1);
        expect(subscription.closed).toBeTrue();
        expect(result).toEqual(["start", 0]);
    }));

});

describe("runInZone/runOutsideZone", () => {

    it("should leave and enter the zone", fakeAsync(() => {
        const zone = {
            run: (fn: () => any) => fn(),
            runOutsideAngular: (fn: () => any) => fn()
        } as NgZone;

        const runSpy = spyOn(zone, "run").and.callThrough();
        const runOutsideAngularSpy = spyOn(zone, "runOutsideAngular").and.callThrough();

        const observable$ = concat(
            of(19)
        ).pipe(
            runInZone(zone),
            runOutsideZone(zone)
        );

        const result: any[] = [];
        const subscription = observable$.subscribe((_) => result.push(_));

        expect(subscription.closed).toBeTrue();
        expect(result).toEqual([19]);
        expect(runSpy).toHaveBeenCalled();
        expect(runOutsideAngularSpy).toHaveBeenCalled();
    }));

});