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

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));

它将 return 我的预期输出：

[ 1, 3, 5, 7, 9 ]

然而在我的代码库中，我有不 return 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 拒绝本身。

我如何才能只处理成功的承诺流而忽略使用高地的失败承诺流？我应该如何处理错误事件？

Answer 1

使用_.errors方法：

Extracts errors from a Stream and applies them to an error handler function. Returns a new Stream with the errors removed (unless the error handler chooses to rethrow them using push). Errors can also be transformed and put back onto the Stream as values.

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

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

它会输出：

[ 1, 3, 5, 7, 9 ]

可以对错误和 return 流的自定义值采取行动或重新抛出错误：

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

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

How to handle promise rejection when creating a highland stream from a promise?

javascript

node.js

typescript

promise

highland.js