Polling requests using rjxs and angular

Question

I am trying to create continuous polling using rxjs and angular. Below is an implementation of my requirement.

My app.component template has for instance 2 or more components (same component).

<widget ticker='BTC'></widget>
<widget ticker='ETH'></widget>

In the widget.component I would like to fetch data from API to populate the widget with the ticker information, but the goal is to collect all the tickers and make just one call e.g (api/crypto/BTC,ETH) and return the data to all widget (2 in this case). Each widget will read the data from the response and keep fetching every minute.

Response example:

{ BTC: { name: 'Bitcoin', price: 7000 }, ETH: { name: 'Etherium', price: 200 }}

My widget component:

export class widgetComponent implements OnInit, OnDestroy {
  @Input() ticker: any;
  subscription: any;

  constructor(
    private cryptoService: CryptoService
  ) { }

  ngOnInit() {
    this.subscription = this.cryptoService
      .setupSymbol(this.ticker)
      .subscribe(data => {
        this.info = data[this.ticker];
      });
  }

  ngOnDestroy() {
    this.subscription.unsubscribe();
  }
}

My Service:

@Injectable({
  providedIn: 'root'
})
export class CryptoService {
  tickers: any = '';
  polledBitcoin$: Observable<number>;
  load$ = new BehaviorSubject('');

  constructor(
    private http: HttpClient
  ) { }

  bitcoin$ = this.http.get(`api/crypto/${this.tickers}`);

  whenToRefresh$ = of('').pipe(
    delay(1000),
    tap(_ => this.load$.next('')),
    skip(1),
  );

  poll$ = concat(this.bitcoin$, this.whenToRefresh$);

  setupTicker(ticker) {
    this.tickers += ticker + ',' ;

    return this.load$.pipe(
      concatMap(_ => this.poll$),
      share()
    );
  }

My code doesn't work as I expect. Each widget makes their own call to the API for there ticker. But I would like just make one call collection all ticker and share the data request through all widgets.

It's not possible to make a stream with rxjs over the one array that contains all tickers e.g ['BTC', 'ETH] and after that start the polling? The polling should wait until all widgets make the setupTicker.

Anyone call help? Thanks in advance.

Answer 1

Have a tickers object that tracks how many widgets are tracking each symbol, each time you add a new symbol start a new polling subscription.

@Injectable({
  providedIn: 'root'
})
export class CryptoService {
  private tickers: { [ticker]: number } = {};
  private subscription: Subscription;

  tickers$ = new BehaviorSubject<{ [ticker]: { name: string, price: number } }>(undefined);

  constructor(
    private http: HttpClient
  ) { }

  subscribe(ticker: string) {
    if (this.tickers[ticker]) {
      this.tickers[ticker]++;
    } else {
      this.tickers[ticker] = 1;
      if (this.subscription) {
        this.subscription.unsubscribe();
      }
      this.subscription = interval(60000).pipe(
        switchMap(() => this.http.get<{ [ticker]: { name: string, price: number } }>(`api/crypto/${Object.keys(this.tickers).join(',')}`))
      ).subscribe(this.tickers$);
    }
  }

  unsubscribe(ticker: string) {
    if (this.tickers[ticker] > 1) {
      this.tickers[ticker]--;
    } else {
      delete this.tickers[ticker];
      if (Object.keys(this.tickers).length === 0) {
        this.subscription.unsubscribe();
      }
    }
  }
}

and in the component

export class widgetComponent implements OnInit, OnDestroy {
  @Input() ticker: string;

  ticker$ = this.cryptoService.tickers$.pipe(
    map(ticker => ticker && ticker[this.ticker])
  );

  constructor(
    private cryptoService: CryptoService
  ) { }

  ngOnInit() {
    this.cryptoService.subscribe(this.ticker);
  }

  ngOnDestroy() {
    this.cryptoService.unsubscribe(this.ticker);
  }
}

and use the async pipe in the template

<ng-content *ngIf="ticker$ | async as tickerVal">
  {{ tickerVal.name }} current price is {{ tickerVal.price }}
</ng-content>