ActivitySubject에 가입했을 때 관찰 가능한 약속을 두 번 실행하지 않는 방법

ActivitySubject에 가입했을 때 관찰 가능한 약속을 두 번 실행하지 않는 방법

나는 다음과 같은 클래스 스토어를 가지고 있다.

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를 구독하는 중get두 개의 다른 구성 요소에서 두 번 관측할 수 있으며getActiveInstallationUseCase 두 번getActiveInstallationUseCase.execute()약속을 이행하다내가 하고 싶은 것은 아무리 가입자가 많아도 사용자가 로그인할 때마다 유스케이스만 운영하는 것이다.

나는 그것을 시도했다.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()사건을 단 한 번 내보내고 대신하려고 노력했다.share와 함께shareReplay하지만 운이 따르지 않았지

---

나는 여기서 그 문제를 반복했다.약속은 4번이고, 나는 2번만 실행했으면 좋겠어.추가하기share()운영자가 작동하게 하지만 내 코드는 그렇지 않다. 왜일까?

rxjs take opprator를 사용해 보십시오.

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

좋아, RxJS에 대해 더 많이 알게 된 후 구독을 공유하는 방법에 대해 잘못 알았다.문제는 다음과 같은 암호에 있었다.

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

이 실행은 관찰 가능한 새로운 것을 반환하는 것이다.An 비록 내가 가지고 있는 모든 사용 사례를 공유할 수 있는 방법이 있지만, 공유는 전혀 진행되지 않았다. 그때마다 나는 공유가 이루어졌기 때문이다..execute()그것은 관찰할 수 있는 새로운 것을 반환했다.

내가 결국 하게 된 것은 관찰할 수 있는 저장고를 만드는 것이었다.나의 모든 사용 사례가 동일한 클래스를 상속받기 때문에 나는 책임의 사슬을 세웠다.만약 그 특별한 관찰이 그 전에 실행되었다면 그것은 공유된다.

기본 사용 사례 클래스:

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.그리고 구조를 개선하는 방법에 대한 어떤 권고사항도 크게 인정받지 못한다!

참조URL: https://stackoverflow.com/questions/58486200/how-to-avoid-executing-a-promise-inside-an-observable-twice-when-subscribed-to-a