正确使用 takeUntil() 来停止 observable答案

【问题标题】：Correctly use takeUntil() for stopping observable正确使用 takeUntil() 来停止 observable
【发布时间】：2017-09-05 20:52:04
【问题描述】：

我正在尝试创建 2 个 observables，它们基本上执行逆运算。它是一个服务发现端点，因此在启动应用程序时，它必须尝试注册到服务发现，直到成功。所以我想创建一个像这样的 observable：

const create$ = Rx.Observable.create((observer) => {
    observer.next('Trying to create observation');
    sp.put(endpoint, { json: true }, (err, res, payload) => {
      err ? observer.error(err) : observer.complete();
    });
  });

当应用程序进行正常关闭时，我想做反向操作。像这样：

const delete$ = Rx.Observable.create((observer) => {
    console.log('deleted subscribed');
    observer.next('Trying to delete observation');
    sp.delete(endpoint, { json: true }, (err, res, payload) => {
      err ? observer.error(err) : observer.complete();
    });
  });

所以我决定创建一个函数，它返回一个具有.create() 和.delete() 的对象。我要解决的问题是，当应用程序启动并尝试注册，但无法到达服务发现端点并且稍后应用程序启动它的正常关闭过程并调用.delete() 操作时，.create() 操作不应该再跑吧。

function observe({ url, version, serviceName }) {
  const endpoint = `/endpoint/${serviceName}/${version}/${encodeURIComponent(url)}`;

  const create$ = Rx.Observable.create((observer) => {
    observer.next('Trying to create observation');
    sp.put(endpoint, { json: true }, (err, res, payload) => {
      err ? observer.error(err) : observer.complete();
    });
  });

  const delete$ = Rx.Observable.create((observer) => {
    console.log('deleted subscribed');
    observer.next('Trying to delete observation');
    sp.delete(endpoint, { json: true }, (err, res, payload) => {
      err ? observer.error(err) : observer.complete();
    });
  });

  return {
    create() {
      return create$.retry(Number.POSITIVE_INFINITY).takeUntil(delete$); // This is where I would want to takeUntil()
    },
    delete({ interval = 5000, times = 0 } = {}) {
      return delete$.retry(times);
    },
  }
}

我遇到的问题是，当使用.takeUntil() 时，它订阅了delete$，它开始执行删除操作，其效果是立即停止create$ observable。

我尝试做takeUntil(Observable.merge(Observable.never(), delete$))，但它订阅了两者，所以不起作用。我也尝试过takeUntil(Observable.concat(Observable.never(), delete$)) 第一个永远不会结束（：P），第二个永远不会订阅。

【问题讨论】：

您的重试逻辑是否正确？描述听起来像是“创建”是重试的那个？
@jayphelps 我更新了创建时的重试逻辑。我想在创建时无限重复，但如果删除开始尝试就停止

标签： rxjs observable observer-pattern rxjs5

【解决方案1】：

大多数 Observable 是冷和单播。这是一个有点冗长的话题，所以我将大部分时间推迟到伟大的文章Hot vs Cold Observables，但总结一下：

COLD 是你的 observable 创建生产者的时候

// COLD
var cold = new Observable((observer) => {
  var producer = new Producer();
  // have observer listen to producer here
});

HOT 是当你的 observable 关闭生产者时

// HOT
var producer = new Producer();
var hot = new Observable((observer) => {
  // have observer listen to producer here
});

在您的示例中，这是一个关键的区别，因为 create$ 和 delete$ 都是冷的。如您所见，向takeUntil 提供delete$ 会导致订阅delete$，从而启动请求。

如果您想保持代码的结构/API 不变，实现此目的的一种方法是使用被视为一种“通知程序”的主题。主题是多播和“热”的（即使它们自己不做任何事情）。

function observe({ url, version, serviceName }) {
  // etc...

  const shutdown$ = new Subject(); // <---------------- create our notifier

  return {
    create() {
      return create$
        .retry(Number.POSITIVE_INFINITY)
        .takeUntil(shutdown$); // <-------------------- take our notifier
    },
    delete({ interval = 5000, times = 0 } = {}) {
      return Observable.defer(() => {
        shutdown$.next(); // <------------------------- notify
        return delete$.retry(times);
      });
    }
  };
}

我们使用了Observable.defer()，以便在有人实际订阅我们返回的 Observable 时执行shutdown$.next() 的副作用。

【讨论】：

【解决方案2】：

方法#1

您可以在 .takeUntill 中使用 Observable，而无需通过创建中间 Subject 来实际订阅它。然后您可以订阅该主题而不是原始主题。像这样的东西：

const delete$ = new Subject();
...
create() {
  return create$
    .takeUntill(delete$);   
},
delete() {
  create(...).subscribe(delete$);
  return delete;
}

方法 #2

但是，在您的情况下，我认为创建主题会更好，该主题将在调用 .delete 时通知。像这样：

const onDelete$ = new Subject();
...
create() {
  return create$
    .takeUntill(onDelete$); 
},
delete() {
  onDelete$.next();
  return ...
}

【讨论】：