Angular RxJS 在一个主题流中结合了同步、异步和 forkJoin

Question

我有一个相当复杂的反应行为，我想使用 RxJS 来实现，但我还没有找到合适的解决方案。

在 Plnkr 中可以找到的我的 Angular 应用程序中，我有一系列同步、异步和并行异步执行。

1. 首先，用户单击一个按钮，该按钮在我的 RxJS 主题上调用 next userClick$ = new Subject<void>();.

   <button (click)="model.on = !model.on; userClick$.next()"> {{ model.on ? 'Stop' : 'Start' }} </button>

2. 然后我使用getRandomNumber 提供一个随机数
3. 然后我使用getExpNumber 计算该随机数的指数
4. 最后我需要同时运行getFloor和getCeil。

我需要什么？

• 能够在最后两次并行执行#4 中使用forkJoin。
• 能够表达#3 对#2 和#4 对#3 的依赖性：
  • getExpNumber 依赖于 getRandomNumber 并在它之后运行
  • getFloor 与 getCeil 并行运行，两者都依赖于 getExpNumber。
• 能够在我的每个flatMap 中使用流的返回值，目前我只得到最后一个值（我需要使用ngFor 和@987654331 向用户显示随机值（第一个flatMap） @管道。

依赖性：我可以通过检查每个 flatMap 中当前可观察值的类型来实现这一点，但我认为必须有更好的方法。

完整代码：

import { Component, NgModule, VERSION } from '@angular/core';
import { BrowserModule } from '@angular/platform-browser';
import { Observable } from 'rxjs/Observable';
import { Subject } from 'rxjs/Subject';
import 'rxjs/add/operator/filter';
import 'rxjs/add/operator/do';
import 'rxjs/add/operator/switchMap';

@Component({
  selector: 'my-app',
  template: `
    <button (click)="model.on = !model.on; userClick$.next()">{{ model.on ? 'Stop' : 'Start' }}</button>
    <h1>Numbers stream</h1>
    <div *ngFor="let numberValue of (numbers$ | async)">
      <h2>{{ numberValue }}</h2>
    </div>
    <label>{{ model.log }}</label>
  `,
})
export class App {
  model = {
    on: false,
    log: ''
  }
  userClick$ = new Subject<void>();
  numbers$: Observable<number[]> = this.userClick$
    .filter(() => !!this.model.on)
    .do(() => this.model.log = '')
    .switchMap(() => this.getRandomNumber())
    .switchMap((num) => this.getExpNumber(num))
    .switchMap((num) => this.getFloor(num))
    .switchMap((num) => this.getCeil(num));
  constructor() {
  }
  /**
   * Runs after user click
   */
  getRandomNumber(): Observable<number[]> {
    return new Observable<number[]>(observer => {
      this.model.log += ' getRandomNumber';
      observer.next([Math.floor(Math.random() * 10) + 1]);

      return () => {}
    }
  }
  /**
   * Depends on getRandomNumber and run after it
   */
  getExpNumber(n: number): Observable<number[]> {
    return new Observable<number[]>(observer => {
      this.model.log += ' getExpNumber';
      observer.next([Math.floor(Math.exp(n)]);

      return () => {}
    }
  }
  /**
   * Runs in parallel with getCeil
   */
  getFloor(n: number): Observable<number[]> {
    return new Observable<number[]>(observer => {
      this.model.log += ' getFloor';
      observer.next([Math.floor(n)]);

      return () => {}
    }
  }
  /**
   * Runs in parallel with getFloor
   */
  getCeil(n: number): Observable<number[]> {
    return new Observable<number[]>(observer => {
      this.model.log += ' getCeil';
      observer.next([Math.ceil(n)]);

      return () => {}
    }
  }
}
@NgModule({
  imports: [ BrowserModule ],
  declarations: [ App ],
  bootstrap: [ App ]
})
export class AppModule {}

Answer 1

如果您只想获取随机数，您可以只订阅一个获取随机数。

this.getRandomNumber().subscribe((value:number)=>{console.log(value)}) 

注意：当我们写

<div *ngFor="let value of ($numbers |async)>{{value}}</div>

是“喜欢”

//in .html
<div *ngFor="let value of numberList>{{value}}</div>
//in .ts
numberList:number[]
subscription:Subscription;

ngOnInit()
{
  this.subscription=$numbers.subscribe(values:number[]=>{
      this.numberList=values;
      this.subscription.unsubscribe()
  }
}

Answer 2

你创建可观察对象的方式对我来说很奇怪（我将使用“of”），我会收到数字，而不是数组，所以代码就像

@Component({
  selector: 'my-app',
  template: `
    <button (click)="model.on = !model.on; userClick$.next()">{{ model.on ? 'Stop' : 'Start' }}</button>
    <h1>Numbers stream</h1>
    <h2 *ngFor="let numberValue of (numbers$ |async)">{{numberValue}}</h2>
    <label>{{ model.log }}</label>
  `,
})
export class HomeComponent {
  model = {
    on: false,
    log: ''
  }
  userClick$ = new Subject<void>();
  numbers$: Observable<number[]> = this.userClick$
    .switchMap(() => {
      return this.getRandomNumber().switchMap((num: number) => {
        this.model.log += num;
        return this.getExpNumber(num).switchMap((num2: number) => {
          this.model.log += num2;
          return forkJoin(this.getFloor(num2), this.getCeil(num2))
        })
      })
    })


  getRandomNumber(): Observable<number> {
    this.model.log += ' getRandomNumber';
    return of((Math.random() * 10) + 1);
  }
  getExpNumber(n: number): Observable<number> {
    this.model.log += ' getExpNumber';
    return of(Math.exp(n));
  }
  getFloor(n: number): Observable<number> {
    this.model.log += ' getFloor';
    return of((Math.floor(n)));
  }

  getCeil(n: number): Observable<number> {
    this.model.log += ' getCeil';
    return of(Math.ceil(n));
  }
}

1. 不使用“do”，do 用于检查响应
2. 如果我像你一样制作 observable，forkJoin 对我不起作用（调用函数，但我无法订阅它们）