rxjs：如何在可观察流中进行条件判断答案

【问题标题】：rxjs: how to do conditional judgement in observable flowrxjs：如何在可观察流中进行条件判断
【发布时间】：2020-02-14 04:06:14
【问题描述】：

在我的 Angular 项目中，我遇到了如下一种情况：
在前端我有一个语言系统，用户可以从选项中选择一种语言。每次用户选择一种新语言时，我都需要用它调用一个服务器端方法subscribe，如果调用成功，它将返回一个subscriptionid。
下次，当用户更改新语言时，我需要创建一个新订阅。但在此之前，我需要unsubscribe当前的。演示代码如下：

  this.start().subscribe((res) => {
    // store current subscription id   
    this.currentSubscriptionId = res.SubscriptionId;
  });

  private start(): Observable<{ [key: string]: string }> {
      return this.xService.language$.pipe(
          switchMap((language) =>
            this.unsubscribe().pipe(
              switchMap((_) => this.invoke("subscribe",language)) // this.invoke return observable<any>
            )
          )
      );
  }

  private unsubscribe(): Observable<any> {
    if (this.currentSubscriptionId) { // default initial value is empty string  ""
      return this.invoke("Unsubscribe", {
        Id: this.currentSubscriptionId
      });
    } else {
      return of("");
    }
  }

所以我为此流程使用状态变量currentSubscriptionId。它的默认值是空字符串，所以第一次不需要调用unsubscribe方法，返回一个of observable。

在当前的实现中，我封装了unsubscribe函数内部的逻辑，并基于状态变量currentSubscriptionId来处理流程。

对于这种情况还有其他优化的解决方案吗？以更 RxJS 的风格来做这种判断控制。

【问题讨论】：

标签： rxjs flow-control

【解决方案1】：

我认为是时候使用“finalize”运算符了，它会在订阅被释放时调用给定的函数。

const { Subscription, BehaviorSubject, Observable, of, never, interval } = rxjs;

const Rx = rxjs.operators;

let cleanUp = new Subscription();

function start() {
  return selectedLang.pipe(
    Rx.switchMap(lang => callSubscribeAPI(lang)),
    Rx.switchMap(id =>
      never().pipe(
        Rx.startWith(id),
        Rx.finalize(() => cleanUp.add(callUnsubscribeAPI(id).subscribe()))
      )
    )
  )
}

// Mock of language selection UI:

let selectedLang = new BehaviorSubject("ja");

// Server-side API mocks:

let subscriptionIdCounter = 0;

function callSubscribeAPI(lang) {
  let id = ++ subscriptionIdCounter;
  console.log(`[server] subscribe: ${lang}@${id}`);
  return of(`${id}`);
}

function callUnsubscribeAPI(id) {
  console.log(`[server] unsubscribe: ${id}`);
  return of();
}

// Usage:

start().subscribe(console.log);

interval(300)
 .pipe(Rx.take(4))
 .subscribe(i => 
   selectedLang.next(["en", "kr", "cn", "ja"][i])
 );

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.4/rxjs.umd.min.js"&gt;&lt;/script&gt;

【讨论】：