从承诺创建高地流时如何处理承诺拒绝？

Question

我通过打字稿在 node@8.11.1 上使用 highland@2.13.0。鉴于此代码片段：

import * as highland from "highland";
import * as lodash from "lodash/fp";

const range = lodash.range(0, 10);
const createPromise = async (i: number): Promise<number> => {
    if (i % 2 !== 0) {
        return Promise.resolve(i);
    }
    return Promise.resolve(null);
};

highland(range).map((i) => {
        return highland(createPromise(i));
    })
    .flatten() // resolving the promises
    .compact() // removing the null values
    .toArray((items) => console.log(items));

它将返回我的预期输出：

[ 1, 3, 5, 7, 9 ]

然而，在我的代码库中，我有一些不返回 null 值但会拒绝该承诺的承诺。但在这种情况下，高原会崩溃：

const createPromise = async (i: number): Promise<number> => {
    if (i % 2 !== 0) {
        return Promise.resolve(i);
    }
    return Promise.reject("Some rejection message");
};


highland(range).map((i) => {
        return highland(createPromise(i));
    })
    .flatten()
    .toArray((items) => console.log(items));

将抛出：

events.js:188
      throw err;
      ^

Error: Unhandled "error" event. (Invalid)
    at Stream.emit (events.js:186:19)
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1908:18
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1593:9
    at Stream.s._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1549:9)
    at Stream.write (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1647:18)
    at Stream._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:974:26)
    at push (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1515:19)
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:3918:13
    at Stream.s._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1549:9)
    at Stream.write (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1647:18)
    at Stream._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:974:26)
    at push (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1515:19)
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:2458:13
    at Stream.s._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1549:9)
    at Stream.write (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1647:18)
    at Stream._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:974:26)
    at Stream.write (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1647:18)
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:680:15
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:3606:17
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1593:9
    at Stream.s._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1549:9)
    at Stream.write (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1647:18)
    at Stream._send (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:974:26)
    at Stream.write (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:1647:18)
    at /home/philipp/rate-pipeline/node_modules/highland/lib/index.js:680:15
    at Immediate._onImmediate (/home/philipp/rate-pipeline/node_modules/highland/lib/index.js:541:17)
    at runCallback (timers.js:794:20)
    at tryOnImmediate (timers.js:752:5)
    at processImmediate [as _immediateCallback] (timers.js:729:5)

我知道我可以将我的 Promise 拒绝转换为 null 值并 compact 它们作为一种解决方法，但我宁愿处理 Promise 拒绝本身。

我怎样才能只处理成功的 Promise 流而忽略使用 highland 的失败？我应该如何处理错误事件？

Answer 1

使用_.errors 方法：

从 Stream 中提取错误并将它们应用于错误处理函数。返回一个删除了错误的新 Stream（除非错误处理程序选择使用 push 重新抛出它们）。错误也可以被转换并作为值放回 Stream。

对于您的用例，这是最少需要的实现：

highland(range).map((i) => {
    return highland(createPromise(i));
  })
  .flatten()
  .errors(() => {})
  .toArray((items) => console.log(items));

它会输出：

[ 1, 3, 5, 7, 9 ]

可以对错误采取行动并将自定义值返回到流或重新引发错误：

.errors((error, push) => {
   if(error.foo === "bar") {
     push(null, null); // pushes null to the result stream
   } else {
     push(err); // re-throws
   }
})