How to avoid executing a promise inside an observable twice when subscribed to a BehaviourSubject?

user_6475656 0
Viewed 296

I have a class Store which is as follows: 

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
}

And my 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 }))))
      )
  }
}

The InstallationStore is being subscribed to the get observable two times in two different components which trigger the getActiveInstallationUseCase twice. getActiveInstallationUseCase.execute() returns a Promise. What I would like to do is that no matter how many subscribers it has, it only runs the use case whenever the user logs in.

I have tried the share() operator with no success as follows:

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

And

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

      )
  }

But it still runs twice. I have checked that this.loginStore.get() emits an event only once and have tried to replace share with shareReplay but with no luck.

I have replicated the issue here. It's calling the promise 4 times, while I would like it to be executed only twice. Adding the share() operator makes it work, however in my code is not, why?

rxjs angular typescript javascript
2 Answers

Try using rxjs take oprator something like

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

Ok, after learning more about RxJS I had a misconception with how to share a subscription. The problem was in this bit of code: 

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

This execute what is doing is returning a new observable. An even though I have a a way of sharing all use cases that I have there wasn't any sharing going on, because each time I did a .execute() it returned a new observable.

What I end up doing was creating a cache of observables. As all my use cases inherit the same class I set up a chain of responsibility. If that particular observable has been executed before then it's shared.

This is the base use case class:

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>
  }
}

Here is an use case:

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))
  }
}

This is the runner:

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!
  }
}

The cache of observables which is implemented as a link of the chain:

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)!
  }
}

And here is how I share the observables, using the 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)
  }
}

All this code can be found in this repository: https://github.com/cesalberca/who-am-i. And any recommendations on how to improve the structure is greatly apreciated!

user_64756560
Related