How to retry with delay an Observable inside pipe in RxJS?

I have the following code:

notificationsWsSubject.pipe(
    filter((socket): socket is Socket => !!socket),
    switchMap(socket => fromEvent<Socket.DisconnectReason>(socket, 'disconnect')),
    tap(() => wsConnectedSubject.next(false)),
    filter(reason => (['ping timeout', 'transport close', 'transport error'] as Socket.DisconnectReason[]).includes(reason)),
    switchMap(() => signedInObservable),
    switchMap(user => forkJoin([of(user), from(getNotificationsWebsocketTicket())])),
).subscribe(values => {
    // Connect with websocket
}, error  => {
    // Throw error to user
})

The general flow:

• Listen to the disconnect event in socket.io-client
• Proceed if the disconnection cause is due to network error
• Get the currently signed in user from signedInObservable
• Generate a ticket in the server by calling getNotificationsWebsocketTicket() (and creating an Observable from it using from()
• Do stuff in subscribe(value => ... )

My problem is that I would like to retry the from(getNotificationsWebsocketTicket()) in case it fails, with a delay of 5 seconds between each failure.
Only after 3 retries I want the entire main observable to fail.

something like:

from(getNotificationsWebsocketTicket()).pipe(delayedRetry(3, 5000))

Is that possible?

Via Active questions tagged javascript - Stack Overflow https://ift.tt/2FdjaAW