如何将 combineLatest 与订阅者一起使用

Question

我想将 combineLatest 与订阅者一起使用，并且仅当传递的 ID 与 itemId 匹配时才通过此方法进行订阅。

这是我到目前为止所做的：

export interface Instance {
  onEvent<T = any>(eventName: string): Subscriber<T>;
}

export interface Payload {
  id: string;
  quantity: number;
}

public async getSingletonSubscription(id: string): Promise<Observable<any>>{
  const instance = await this.instance;
  // itemids is an Observable coming from another method. Just did an array for this example
  const itemIds: Array<number> = [1, 2];

  // Subscriber<any>
  const subscriber = instance.onEvent(message);

  const observable = combineLatest(
    subscriber,
    itemIds
  ).pipe(
    filter((payload: any, ids: any) => ids.include(id))
  );

  return observable
}

我目前遇到 Argument of type 'Subscriber<any>' is not assignable to parameter of type 'ObservableInput<any> | SchedulerLike | ((...values: any[]) => unknown)' 类型错误，我目前不确定如何解决此类型错误。

通常我会这样使用interface.onEvent

this.interface
  .onEvent(message)
  .subscribe((payload: Payload)) => console.log(payload));

但是因为 itemIds 也是可观察的。当我 return 一个 observable

时，我想使用 combineLatest 订阅一次

Answer 1

使用 any 作为函数的 return 类型而不是 Promise。希望您能找到解决方案。

Answer 2

你的代码示例中有很多错误陈述，但我会尽力解释每一个。

基于以下内容：

this.interface
  .onEvent(message)
  .subscribe((payload: Payload)) => console.log(payload));

this.interface.onEvent() 的 return 类型是 Observable，而不是 Subscriber。 Subscriber 是具有活动 Subscription 到 Observable.

的东西

关于你的方法，我会避免混淆 Promises 和 Observables。如果你正在使用 RxJS，那么让一切都成为 Observable 会更容易。由于 this.interface.onEvent(message) return 是 Observable，我们可以保持原样。

现在更新代码：

export interface Instance {
  onEvent<T>(eventName: string): Observable<T>;
}

export interface Payload {
  id: string;
  quantity: number;
}

public getSingletonObservable(id: string): Observable<Payload>{
  const instance$ = this.instance.onEvent<Payload>(message);
  // To make this an observable, we need to wrap it inside the `of()` operator.
  const itemIds$ = of([1, 2]);

  const observable$ = combineLatest([
    subscriber,
    itemIds
  ]).pipe(
    filter(([payload, ids]) => ids.include(payload.id)),
    // I'm assuming we want to return the payload if it passes the previous condition.
    map(([payload, ids]) => payload)
  );

  return observable$;
}

我更新了打字所以我们不使用 any（在 TypeScript 中应该不惜一切代价避免）。

我还更新了一些变量名，因为如果它是可观察的，通常将 $ 添加到名称的末尾。

您还忘记声明 id 的定义位置，所以我添加了引用，它是 payload 的属性。

如何将 combineLatest 与订阅者一起使用

how to use combineLatest with a Subscriber

rxjs

angular

rxjs-pipeable-operators

rxjs6

rxjs-observables