使用 rjxs 和 angular 轮询请求答案

【问题标题】：Polling requests using rjxs and angular使用 rjxs 和 angular 轮询请求
【发布时间】：2019-05-12 19:54:41
【问题描述】：

我正在尝试使用 rxjs 和 angular 创建连续轮询。以下是我的要求的实现。

我的 app.component 模板有例如 2 个或更多组件（相同的组件）。

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

在 widget.component 中，我想从 API 获取数据以使用代码信息填充小部件，但目标是收集所有代码并只进行一次调用，例如 (api/crypto/BTC,ETH) 和将数据返回给所有小部件（在本例中为 2）。每个小部件都会从响应中读取数据并继续每分钟获取。

响应示例：

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

我的小部件组件：

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

我的服务：

@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()
    );
  }

我的代码没有按预期工作。每个小部件都会对 API 进行自己的调用以获取代码。但我只想让一个呼叫集合全部显示并通过所有小部件共享数据请求。

不可能在包含所有代码的一个数组上使用 rxjs 创建一个流，例如 ['BTC', 'ETH] ，然后开始轮询？轮询应该等到所有小部件都完成 setupTicker。

有人求助吗？提前致谢。

【问题讨论】：

您只需要收集您服务中的所有代码，然后调用API。由于小部件组件不是动态的，您可以对其计数进行硬编码以了解所有代码何时都存在
小部件组件是动态的。这个想法实际上是在页面中添加/删除（拖放）更多小部件。
小部件都应该引用一个公共服务。当他们向该服务询问其代码时，该服务应更新其所有票证列表并返回组合流
感谢您的回答，@adrian 提供的以下代码就像一个魅力！

标签： angular rxjs polling

【解决方案1】：

有一个代码对象来跟踪跟踪每个交易品种的小部件数量，每次添加新交易品种时都会开始新的轮询订阅。

@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();
      }
    }
  }
}

在组件中

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

并在模板中使用异步管道

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

【讨论】：