订阅 BehaviorSubject 时，如何避免在可观察对象内执行两次承诺？

Question

我有一个类商店如下：

import { BehaviorSubject, Observable } from 'rxjs'

export abstract class Store<T> {
  private state: BehaviorSubject<T> = new BehaviorSubject((undefined as unknown) as T)

  get(): Observable<T> {
    return this.state.asObservable()
  }

  set(nextState: T) {
    return this.state.next(nextState)
  }

  value() {
    return this.state.getValue()
  }

  patch(params: Partial<T>) {
    this.set({ ...this.value(), ...params })
  }

  abstract create(): void
}

还有我的 InstallationStore：

import { Store } from '../../store/store'
import { Installation } from '../domain/installation/installation'
import { from, Observable } from 'rxjs'
import { GetActiveInstallationUseCase } from '../../../features/planning/application/get-active-installation-use-case'
import { Injectable } from '@angular/core'
import { map, switchMap } from 'rxjs/operators'
import { LoginStore } from '../../../features/login/application/login-store'

interface State {
  activeInstallation: Installation
}

@Injectable({
  providedIn: 'root'
})
export class InstallationStore extends Store<State> {
  constructor(
    private readonly getActiveInstallationUseCase: GetActiveInstallationUseCase,
    private readonly loginStore: LoginStore
  ) {
    super()
    this.create()
  }

  create(): void {
    this.set({
      activeInstallation: {
        isDefault: true,
        productionProfile: 'baz',
        incomingProfile: 'foo',
        id: 1,
        energeticRole: 'bar',
        name: ''
      }
    })
  }

  get(): Observable<State> {
    return this.loginStore
      .get()
      .pipe(
        switchMap(() => from(this.getActiveInstallationUseCase.execute()).pipe(map(x => ({ activeInstallation: x }))))
      )
  }
}

InstallationStore 在触发getActiveInstallationUseCase 两次的两个不同组件中被订阅get observable 两次。 getActiveInstallationUseCase.execute() 返回一个 Promise。我想做的是无论它有多少订阅者，它只会在用户登录时运行用例。

我试过share() 操作符没有成功，如下：

get(): Observable<State> {
    return this.loginStore
      .get()
      .pipe(
        switchMap(() => from(this.getActiveInstallationUseCase.execute()).pipe(map(x => ({ activeInstallation: x })))),
        share()
      )
  }

和

get(): Observable<State> {
    return this.loginStore
      .get()
      .pipe(
        switchMap(() => from(this.getActiveInstallationUseCase.execute()).pipe(map(x => ({ activeInstallation: x }))), share()),

      )
  }

但它仍然运行两次。我检查了this.loginStore.get() 只发出一次事件，并尝试用shareReplay 替换share，但没有成功。

---

我已经复制了问题here。它调用了承诺 4 次，而我希望它只执行两次。添加share() 运算符使其工作，但在我的代码中没有，为什么？

Answer 1

尝试使用 rxjs take 类似的操作符

 get(): Observable<State> {
    return this.loginStore
      .get()
      .pipe(
        take(1),
        switchMap(() => from(this.getActiveInstallationUseCase.execute()).pipe(map(x => ({ activeInstallation: x }))))
      )
  }

Answer 2

好的，在了解了 RxJS 的更多信息后，我对如何共享订阅产生了误解。问题出在这段代码中：

get(): Observable<State> {
    return this.loginStore
      .get()
      .pipe(
        switchMap(() => from(this.getActiveInstallationUseCase.execute() /* HERE */).pipe(map(x => ({ activeInstallation: x }))))
      )
  }

这个执行正在做的是返回一个新的 observable。即使我有一种分享所有用例的方法，我也没有进行任何分享，因为每次我执行 .execute() 时，它都会返回一个新的 observable。

我最终做的是创建一个可观察对象的缓存。因为我所有的用例都继承了同一个类，所以我设置了一个chain of responsibility。如果那个特定的 observable 之前已经执行过，那么它就是共享的。

这是基本用例类：

import { Observable } from 'rxjs'
import { dependencyTree } from '../../dependency-tree'

export abstract class UseCase<Param, Result> {
  abstract readonly: boolean

  abstract internalExecute(param: Param): Observable<Result>

  execute(param: Param): Observable<Result> {
    const runner = dependencyTree.runner
    return runner.run(this, param) as Observable<Result>
  }
}

这是一个用例：

import { Observable } from 'rxjs'
import { GameRepository } from '../domain/game-repository'
import { Id } from '../../../core/id'
import { map } from 'rxjs/operators'
import { Query } from '../../../core/use-case/query'

type Params = { id: Id }

export class HasGameStartedQry extends Query<boolean, Params> {
  constructor(private readonly gameRepository: GameRepository) {
    super()
  }

  internalExecute({ id }: Params): Observable<boolean> {
    return this.gameRepository.find(id).pipe(map(x => x?.start !== undefined ?? false))
  }
}

这是赛跑者：

import { ExecutorLink } from './links/executor-link'
import { Observable } from 'rxjs'
import { LoggerLink } from './links/logger-link'
import { Context } from './context'
import { UseCase } from './use-case'
import { CacheLink } from './links/cache-link'

export class Runner {
  chain = this.cacheLink.setNext(this.executorLink.setNext(this.loggerLink))

  constructor(
    private readonly executorLink: ExecutorLink,
    private readonly loggerLink: LoggerLink,
    private readonly cacheLink: CacheLink
  ) {}

  run(useCase: UseCase<unknown, unknown>, param?: unknown): Observable<unknown> {
    const context = Context.create({ useCase, param })
    this.chain.next(context)
    return context.observable!
  }
}

被实现为链的链接的可观察对象的缓存：

import { BaseLink } from './base-link'
import { Context } from '../context'
import { Observable } from 'rxjs'

export class CacheLink extends BaseLink {
  private readonly cache = new Map<string, Observable<unknown>>()

  next(context: Context): void {
    if (context.param !== undefined) {
      this.nextLink.next(context)
      return
    }

    if (!this.cache.has(context.useCase.constructor.name)) {
      this.nextLink.next(context)
      this.cache.set(context.useCase.constructor.name, context.observable)
    }

    context.observable = this.cache.get(context.useCase.constructor.name)!
  }
}

以下是我使用ExecutorLink 分享可观察数据的方式：

import { BaseLink } from './base-link'
import { Context } from '../context'
import { share } from 'rxjs/operators'

export class ExecutorLink extends BaseLink {
  next(context: Context): void {
    if (!context.hasSetObservable) {
      const observable = context.useCase.internalExecute(context.param)
      if (context.useCase.readonly) {
        context.observable = observable.pipe(share())
      } else {
        context.observable = observable
      }
    }
    this.nextLink.next(context)
  }
}

所有这些代码都可以在这个存储库中找到：https://github.com/cesalberca/who-am-i。非常感谢任何有关如何改进结构的建议！