nga-29

.ts

export function grpcStream<T>(client: ClientReadableStream<T>): Observable<T> {
    let stream: ClientReadableStream<T> = null;
    let subscriptionCounter = 0;

    const data: Observable<any> = new Observable((observer: Observer<T>) => {
        if (subscriptionCounter === 0) {
            stream = client;
        }
        subscriptionCounter++;

        stream.on(StreamType.DATA, (response: jspb.Message) => {
            observer.next(response.toObject());
        });

        stream.on(StreamType.STATUS, (status: Status) => {
            if (status.code === StatusCode.UNAUTHENTICATED) {
                jwtAuthError$.next();
            }

            if (status.code !== StatusCode.OK) {
                observer.error(status);
            }
        });
    });

    return data.pipe(
        finalize(() => {
            subscriptionCounter--;

            if (subscriptionCounter === 0) {
                stream.cancel();
            }
        }),
        share(),
        retryWhen(errors =>
            errors.pipe(
                tap(val => console.warn(`Stream will be reconnected in 30 seconds`)),
                delayWhen(val => timer(30000)),
            ),
        ),
    );
}