src/app/util/rxjs/rxjs.util.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 {ofType} from "@ngrx/effects";
import {Action, ActionCreator, Creator} from "@ngrx/store";
import {concat, EMPTY, Observable, ObservableInput, of, OperatorFunction, pipe, throwError, timer} from "rxjs";
import {catchError, concatMap, endWith, map, retryWhen, switchMap, toArray} from "rxjs/operators";

/**
 * If an error occurs, the observable is repeated after a given amount of time.
 * @param timeInMs Time in miliseconds after which the observable is repeated.
 * @param count Number of times the observable shall be repeated; if not specified, the observable is repeated until completion.
 */
export function retryAfter<T>(timeInMs: number, count?: number) {
    let counter = 0;
    return pipe(
        retryWhen<T>((errors) => {
            return errors.pipe(
                concatMap((error) => {
                    if (typeof count === "number" && counter > count) {
                        return throwError(error);
                    }
                    counter++;
                    return timer(timeInMs).pipe(map(() => error));
                })
            );
        })
    );
}

/**
 * If an error occurs, error is just ignored and the observable is finished without emitting any values.
 */
export function ignoreError<T>(): OperatorFunction<T, T> {
    return catchError(() => {
        return EMPTY;
    });
}

/**
 * Catches all errors and maps them to the given value.
 */
export function catchErrorTo<T, O = T>(value: O) {
    return catchError<T, ObservableInput<O>>(() => of(value));
}

/**
 * An error is thrown after a specific rxjs action is emitted by the observable.
 */
export function throwAfterActionType<AC extends ActionCreator<string, Creator>[], U extends Action = Action, V = ReturnType<AC[number]>>(
    ...allowedTypes: AC
) {
    return concatMap<U, Observable<U>>((v) => {
        return concat(of(v), of(v).pipe(ofType(...allowedTypes), concatMap(() => throwError(v))));
    });
}

/**
 * The current observables ends with the subscription to another one provided by the given factory function.
 */
export function endWithObservable<T, Z = T>(endWithFactory: () => ObservableInput<Z>): OperatorFunction<T, T | Z> {
    const TOKEN: T = {} as any;
    return pipe(
        endWith<T, T>(TOKEN),
        concatMap<T, ObservableInput<T | Z>>((_) => _ === TOKEN ? endWithFactory() : of(_))
    );
}

/**
 * All emitted values of an observable will be delayed until completion.
 */
export function emitOnComplete<T>() {
    return pipe(
        toArray<T>(),
        switchMap((_) => of<T>(..._))
    );
}

/**
 * Reenters the angular zone for all emitted values.
 */
export function runInZone<T>(zone: NgZone): OperatorFunction<T, T> {
    return (source) => {
        return new Observable<T>(observer => {
            return source.subscribe({
                next: (x) => zone.run(() => observer.next(x)),
                error: (err) => observer.error(err),
                complete: () => observer.complete()
            });
        });
    };
}

/**
 * Leaves the angular zone for all emitted values.
 */
export function runOutsideZone<T>(zone: NgZone): OperatorFunction<T, T> {
    return (source) => {
        return new Observable<T>(observer => {
            return source.subscribe({
                next: (x) => zone.runOutsideAngular(() => observer.next(x)),
                error: (err) => observer.error(err),
                complete: () => observer.complete()
            });
        });
    };
}