前言

在前面，我们已经用 Rxjs 实现了 Store 和 Action。那么，对于前端开发领域非常重要的一块—-数据拉取，又如何用 Rxjs 来实现呢？

异步 Action

在 Redux 中，我们把拉取数据的 Action 叫做异步 Action，主要实现方式有：

• 直接 ajax 请求，在回调中执行其他的 Action
• 使用 redux-thunk，将异步 Action 分解为 Loading、Success、Failure 等状态，分别 dispatch 一个新的 Action
• 使用 redux-saga，将异步操作同步化
• 其他的库

那么，我们在 Rxjs 中，也可以参考 Redux，实现我们的异步操作。

异步请求也是一个流

Rxjs 提供了 Observable.fromPromise 方法，该方法可以将 Promise 函数转化为一个流，在 Promise resolve 数据的时候，emit 该数据，反之在 Promise reject 错误的时候，抛出该错误。
所以，我们将异步请求包装为一个 Promise 函数，并转化为一个流，这里使用了 fetch

const needBody = method =>
  method === 'POST' || method === 'PATCH' || method === 'PUT';

const needQuery = method => method === 'GET';

const fetchData = (method, url, data?, options?) => new Promise((resolve, reject) => {
  let finalURL = url;
  if (needQuery(method) && data) {
    const params = Object.keys(data)
      .filter(key => data[key] || data[key] === false)
      .map(key => encodeURIComponent(key) + '=' + encodeURIComponent(data[key]))
      .join('&')
      .replace(/%20/g, '+');
    finalURL = `${url}?${params}`;
  }

  const finalOptions = Object.assign({}, {
    headers: {
      'Content-Type': 'application/json'
    },
    credentials: 'same-origin',
    method
  }, options);
  if (needBody(method)) {
    finalOptions.body = JSON.stringify(data);
  }

  fetch(finalURL, finalOptions).then(response => {
    if (response.ok) {
      response.json().then(res => resolve(res));
    } else {
      response.text().then(res => reject(res));
    }
  }).catch(err => {
    reject(err);
  });
});

export const http = (method, url, data?, options?, errNotification?) => {
  return Observable.fromPromise<any>(fetchData(method, url, data, options, errNotification));
};

请求结果也是一个流

以上的 http 函数就是一个流，只要被订阅，就会执行异步操作，并且将请求结果或者失败原因用流的形式返回。

http('GET', '/user', {}, {}).subscribe(data => {
  console.log(data);
}, err => {
  console.log(err);
});

这个流订阅一次之后只会执行一次，也符合我们对异步 Action 的需求。
然而，当多次发起请求（多次订阅）后会发现，我们之前的订阅没有被释放，造成了资源的浪费。所以我们定义自己的 request 函数如下：

const request = ({
  method, url, data, options
}) => {
  const sub = http(method, url, data, options).subscribe(data => {
    console.log(data);
    sub.unsubscribe();
  }, err => {
    console.log(err);
    sub.unsubscribe();
  });
};

自定义成功和失败回调

我们需要在请求成功和失败之后，执行一些业务，这个时候就需要成功和失败的回调了。话不多说，直接上代码

const request = ({
  method, url, data, options, sucCallback, errCallback
}) => {
  const sub = http(method, url, data, options).subscribe(data => {
    sucCallback(data);
    sub.unsubscribe();
  }, err => {
    errCallback(err);
    sub.unsubscribe(); 
  });
};

这就是我们最终的请求函数。

异步 Action 的最终形态

有了以上的请求函数，实现我们的异步 Action 就很方便了。比如一个拉取用户列表的 action，实现如下：

export const loadUserList = params => {
  // loading
  userListLoading$.next(true);

  request({
    method: 'GET',
    url: '/user',
    data: params,
    sucCallback: (data) => {
      // 获取到数据，让 store 执行 next
      userList$.next(data);
    },
    errCallback: (err) => {
      // 错误处理
      showErrorToast(err);
    }
  })
};

这样，我们就实现了在 React 中，调用 loadUserList 函数即可获取数据并改变 Store 中的值，UI 层只需要关注渲染，数据层隔离开来单独实现逻辑。

总结

在本文我们实现了异步 Action，加上上文的 Store，普通的 Action，基本上可以实现所有的数据层面的操作了。还在等什么，赶紧用上 Rxjs，体验其独特的魔力。