rxjs:如何在可观察到的流中进行条件判断

Question

在我的Angular项目中，我遇到一个案例，如下所示：

在前端我有一个语言系统，用户可以从选项中选择一种语言。每次用户选择一种新语言时，我都需要使用它调用服务器端方法subscribe，如果调用成功，它将返回到subscriptionid。

下一次，当用户更改新语言时，我需要创建一个新的订阅。但在此之前，我需要unsubscribe当前的。demo代码如下：

  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风格的方式进行这种判断控制。

Answer 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])
 );

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.4/rxjs.umd.min.js"></script>

​