Tips for preventing a promise from being executed more than once within an observable while being subscribed to a BehaviorSubject

Question

Tips for preventing a promise from being executed more than once within an observable while being subscribed to a BehaviorSubject

I've defined a class called Store with the following structure:

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
}

Now let's take a look at 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 problem I'm facing is that the InstallationStore is subscribed to the get observable twice in different components, causing the getActiveInstallationUseCase to run twice. The getActiveInstallationUseCase.execute() method returns a Promise, and I want it to execute only once when the user logs in.

I attempted to use the share() operator without success as shown below:

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

And also tried:

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

      )
  }

Despite these attempts, the method still runs twice. I have verified that this.loginStore.get() emits an event just once and even tried replacing share with shareReplay but couldn't resolve the issue.

I created a demo of the issue here. Currently, the promise is being called 4 times, whereas I want it to be executed only twice. While using the share() operator solves the problem in the demo, it doesn't work as expected in my actual code. Why is that happening?

javascript angular typescript rxjs

Answer 1

Answer №1

Consider incorporating the rxjs take method in your code, like this:

 fetch(): Observable<Data> {
    return this.dataService
      .fetch()
      .pipes(
        take(1),
        switchMap(() => from(this.getDataUseCase.execute()).pipe(map(y => ({ newData: y }))))
      )
  }

Answer 2

Consider incorporating the rxjs take method in your code, like this:

 fetch(): Observable<Data> {
    return this.dataService
      .fetch()
      .pipes(
        take(1),
        switchMap(() => from(this.getDataUseCase.execute()).pipe(map(y => ({ newData: y }))))
      )
  }

Answer 3

Answer №2

Upon delving deeper into RxJS, I discovered a misconception regarding how to share a subscription. The issue lay within this portion of code:

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

The execute function was actually returning a new observable each time it was called. Despite having a way to share all use cases in place, there was no actual sharing occurring due to the creation of new observables with each execution of .execute().

To address this issue, I implemented a cache system for observables. Given that all my use cases inherit from the same class, I established a chain of responsibility. If an observable had been executed before, it would be shared.

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

An example of a 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, implemented as part 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 components of this code can be accessed in this repository: https://github.com/cesalberca/who-am-i. Any suggestions on enhancing the structure are highly welcomed!

Answer 4

Upon delving deeper into RxJS, I discovered a misconception regarding how to share a subscription. The issue lay within this portion of code:

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

The execute function was actually returning a new observable each time it was called. Despite having a way to share all use cases in place, there was no actual sharing occurring due to the creation of new observables with each execution of .execute().

To address this issue, I implemented a cache system for observables. Given that all my use cases inherit from the same class, I established a chain of responsibility. If an observable had been executed before, it would be shared.

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

An example of a 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, implemented as part 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 components of this code can be accessed in this repository: https://github.com/cesalberca/who-am-i. Any suggestions on enhancing the structure are highly welcomed!

Tips for preventing a promise from being executed more than once within an observable while being subscribed to a BehaviorSubject

Answer №1

Answer №2

Similar questions

How to transition from using a CDN to NPM for implementing the Google Maps JavaScript MarkerClusterer?

Using Node.js to send a response only after every promise has been resolved

What steps are needed to switch colors after a loop has been clicked?

Discovering the origins of the node.js native modules and delving into the intricacies of typed modules

What is the process for generating a new object by outputting key-value pairs?

What is the best method for fetching the values of a select element in React.js?

Making AngularJS 'PUT' requests: The process of submitting only the data in a form

PrismaClient is currently incompatible with this browser environment and has been optimized for use in an unknown browser when performing updates

Issue with Videogular: hh:mm:ss Date filter not functioning properly

Troubleshooting the issue with the htmlFor attribute

Is it possible to use a Backbone Model for an unconventional HTTP POST request that isn't

Testing the subscription feature with unit tests

Deleting a key from a type in TypeScript using subtraction type

Having trouble sending a POST request to a Node.js HTTP Webserver unless it's from the localhost

Angular - Ensuring the Quality of Your Code

How can you apply a class to a different element by hovering over one element?

What is the method to execute a prototype function within another prototype?

Problems with Ajax calls are preventing Internet Explorer9 and Internet Explorer10 from functioning properly

Determine the data type of a string without needing to convert it

Transforming Objects with THREE.js: Navigating the Order of Transformations