如何让Observable.flatMap等待解析

Question

在使用 RxJS 和 React 时，我遇到了如何等待另一个 Observable 的地图中生成的 Observable.fromPromise 中的数据的问题。

我有异步助手：

const dataStreamGenerator = (url = CLIENTS_DATA_URL) => (
  Rx.Observable.fromPromise(fetch(url))
  .flatMap(response => Rx.Observable.fromPromise(response.json()))
  .catch(err => Rx.Observable.of(new Error(err)))
);

然后我有 actions.fetchClients$ 即 Rx.Subject:

actions.fetchClients$.map((url = CLIENTS_DATA_URL) => {
    const ts = Date.now();
    console.log('FETCH CLIENTS with: ', url);
    return dataStreamGenerator(url);
  }).map(val => {
    console.log('GOT DATA IN REDUCER: ', val);
    const error = (val instanceof Error) ? val.message : undefined;
    const data = error ? undefined : val;
    actions.receivedClientsData$.next({ data, error, ts: Date.now() });
    return (state) => state;
  })

（是的，试图在 RxJS 中模仿 Redux）。

当我测试 dataStreamGenerator 时，它工作正常（使用 ava）并传送数据：

test('dataStreamGenerator', t => {
  const ds$ = dataStreamGenerator(CLIENTS_DATA_URL);
  return ds$.do((data) => {
    t.is(data.length, 10);
  });
});

（AVA自动订阅observable并消费，无需订阅）

但是 actions.fetchClients$.map((url = CLI... 第二张地图（开始... console.log('GOT DATA IN REDUCER: ', val);... 仍在获取 Observable 而不是来自 dataStream$ 的数据。

我在 fetchClients$ 代码中尝试了 map 和 flatMap 的所有可能组合，但仍然没有成功。

我的测试代码是：

test.only('fetchClients$', t => {
  const initialState = {};
  actions.fetchClients$.next('http://jsonplaceholder.typicode.com/users');
  reducer$.subscribe(val => {
    console.log('TEST SUBSCRIBE VAL: ', val);
    t.pass();
  });
  actions.fetchClients$.next('http://jsonplaceholder.typicode.com/users');
});

我不知道如何等待 Observable dataStreamGenerator(url); 发出数据而不是 Observable。

谢谢。

Answer 1

您需要展平从 dataStreamGenerator 返回的结果。

actions.fetchClients$
  //This is now a flatMap
  .flatMap((url = CLIENTS_DATA_URL) => {
    const ts = Date.now();
    console.log('FETCH CLIENTS with: ', url);
    return dataStreamGenerator(url);
  });

map 运算符按原样向下游传递值 ，而 flatMap 将展平 Observables、Arrays和 Promises 这样它们的值就是传播的值。

它在测试用例中有效，因为您直接订阅了 dataStreamGenerator 返回的 Observable。

如何让Observable.flatMap等待解析

How to make Observable.flatMap to wait to resolve

javascript

rxjs

ava