触发 takeUntil 操作时执行函数

Question

我想在触发 takeUntil 运算符时执行函数（在 ajaxObservable 之外）。 In this way 立即被解雇，因为我调用了 observer.complete()，但如果我不调用它，就会发生内存泄漏，对吗？
实现此目标的好方法是什么？

export const ajaxObservable = (url, method, data, params) => {
    let cancelToken = axios.CancelToke.source()
    return Observable
        .fromPromise(axios({/* axios params */}))
        .map(r => r.data)
        .catch(err => Observable.throw(err))
        .merge(
            new Observable(observer => {
                observer.complete()
                return () => cancelToke.cancel()
            })
        )
 }

Answer 1

您的方法的一个问题是调用 ajaxObservable() 不会 return lazy Observable。相反，ajax require 会立即由 axios 发出，即使没有人订阅 returned Observable。

虽然有例外，但通常最好的做法是让像这样的自定义 Observable 变得懒惰，因此对用户来说同样的期望也适用。

为此，您需要 return 一个新的匿名 Observable，这与您使用 Promises 执行此操作的方式非常相似。在我们的自定义 Observable 订阅处理程序中，没有太多需要使用 fromPromise 或任何 rxjs，因为我们只需要调用 axios 和 then.

作为奖励，当我们这样做时，您原来问题的解决方案变得更加明显：如果有人取消订阅，我们可以调用 cancelToken.cancel()。

export const ajaxObservable = (url, method, data, params) => {
    return new Observable(observer => {
        let cancelTokenSource = axios.CancelToken.source();
        let options = {
          url,
          method,
          data,
          params,
          withCredentials: true,
          cancelToken: cancelTokenSource.token
        };

        axios(options)
            .then(response => {
                observer.next(response.data);
                observer.complete(); // commonly forgotten, but critical!
            }, error => {
                observer.error(error);
            });

        return () => cancelTokenSource.cancel();
    });
};

顺便说一句 .catch(err => Observable.throw(err)) 实际上是一个 noop，再次抛出相同的错误。

您可能有兴趣知道 rxjs 带有 AjaxObservable，这使得 axios 之类的东西变得不必要了。不幸的是，它的文档在 rxjs v5 文档中没有正确显示，但可以在线找到它：http://reactivex.io/rxjs/file/es6/observable/dom/AjaxObservable.js.html 它有一个非常标准的 API，类似于大多数 ajax 实用程序。

/**
 * Creates an observable for an Ajax request with either a request object with
 * url, headers, etc or a string for a URL.
 *
 * @example
 * source = Rx.Observable.ajax('/products');
 * source = Rx.Observable.ajax({ url: 'products', method: 'GET' });
 *
 * @param {string|Object} request Can be one of the following:
 *   A string of the URL to make the Ajax call.
 *   An object with the following properties
 *   - url: URL of the request
 *   - body: The body of the request
 *   - method: Method of the request, such as GET, POST, PUT, PATCH, DELETE
 *   - async: Whether the request is async
 *   - headers: Optional headers
 *   - crossDomain: true if a cross domain request, else false
 *   - createXHR: a function to override if you need to use an alternate
 *   XMLHttpRequest implementation.
 *   - resultSelector: a function to use to alter the output value type of
 *   the Observable. Gets {@link AjaxResponse} as an argument.
 */

它还有简写如ajax.getJSON等

触发 takeUntil 操作时执行函数

execute function when takeUntil operation is triggered

javascript

rxjs

reactjs

rxjs5

redux-observable