concatMap и поделиться, чтобы играть серию для наблюдаемых только один раз, даже если несколько подписчиков

Question

Я пытаюсь выполнить некоторые наблюдаемые последовательно, что я сделал с concatMap, но я хочу, чтобы они выполнялись только один раз, даже если у меня несколько подписчиков, я пробовал операторы share и shareReplay, но ни один из нихкажется, работает.

Вот код:

import { EMPTY, Observable, BehaviorSubject, of, from, concat, timer } from 'rxjs';
import { concatMap, map, switchMap, tap, filter, take, share, shareReplay, exhaust, publishLast, refCount } from 'rxjs/operators';


const refreshAccessToken = () => {
  console.log("Start refresh access token")
  return timer(2000);
}

const connect = () => {
  console.log("Start connect")
  return timer(3000)
}

const authent = () => {
  console.log("Start authent")
  return timer(2000)
}

let isAuthenticated = false;
let isExpired = true;

function initSocket(): Observable<any> {
  if (!isAuthenticated) {
    let observable$ = of(1);
    if (isExpired) {
      observable$ = refreshAccessToken();
    }
    return observable$.pipe(
      concatMap(() => connect()),
      concatMap(() => authent()),
      tap(() => {
        isAuthenticated = true;
        console.log("Tap")
      }),
      share()
    );
  }
  return of(1);
}


initSocket().subscribe(() => console.log("Finished 1"))

initSocket().subscribe(() => console.log("Finished 2"))

Вы можете найти мой образец в https://stackblitz.com/edit/rxjs-4memxw?embed=1&file=index.ts

Вот результаты:

Start refresh access token
Start refresh access token
Start connect
Start connect
Start authent
Start authent
Tap
Finished 1
Tap
Finished 2

Но то, что я ищу, это:

Start refresh access token
Start connect
Start authent
Tap
Finished 1
Finished 2

Возможно, на этот вопрос уже есть ответ, но я не вижу, что мне не хватает, если кто-то может помочь, было бы неплохо.

РЕДАКТИРОВАТЬ:

Я объясню мою потребность, может быть, я решаю проблему неправильно. Я кодирую сервисный класс, который обертывает клиента socket.io, каждая функция в этом классе, которая генерирует или прослушивает событие socket.io, должна проверить, не подключены ли мы и не аутентифицированы ли в socket.io.

Вот что должно быть сделано несколькими наблюдаемыми в каждой подписке:

1. Проверьте, подключено ли
2. Если нет, проверьте, все ли действительный токен доступа, если да, перейдите к 4
3. Если нет, обновите токен доступа
4. Создайте соединение socket.io и дождитесь события соединения
5. При событии соединения отправьте access_token и дождитесь ответа
6. Если аутентификация успешна, переходите к следующей наблюдаемой

В этой серии наблюдаемых первый подписчик выполнит все наблюдаемые, если второй подписчик прибывает, когда наблюдаемая выполняет ожидание того же самого результата, неИнициируйте соединение дважды.

Если третье или больше поступает после завершения соединения, нам все равно нужно проверить, является ли переменная isАутентифицированный изменен, если он ложный, действуйте как первый подписчик, если это правда, продолжайте.

Надеюсь, это поможет уточнить, что я ищу.

Answer 1

Вы хотите присвоить данные наблюдаемой, а затем подписаться на эту наблюдаемую область вместо вызова initFunction(), которая создает столько наблюдаемых, сколько имеется подписчиков. Но назначение данных наблюдаемым и подписка на них предотвращает это, также используя shareReplay:

let obs$ = new Observable();

function initSocket(): void {
  if (!isAuthenticated) {
    let observable$ = of(1);
    if (isExpired) {
      observable$ = refreshAccessToken();
    }
    obs$ = observable$.pipe(
      concatMap(() => connect()),
      concatMap(() => authent()),
      tap(() => {
        isAuthenticated = true;
        console.log("Tap")
      }),
      shareReplay()
    );
  }
}

initSocket();

obs$.subscribe(() => console.log("Finished 1"))

obs$.subscribe(() => console.log("Finished 2"))

setTimeout(() => {
  obs$.subscribe(() => console.log("Finished 3 after 20 seconds"))
}, 20000)

STACKBLITZ