触发 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() 不会返回 lazy Observable。相反，即使没有人订阅返回的 Observable，也会立即由 axios 生成 ajax require。

虽然有例外，但通常最好的做法是让像这样的自定义 Observables 是惰性的，因此对用户也有同样的期望。

为此，您需要返回一个新的匿名 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();
    });
};

---

btw .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 这样的简写形式。