在发布订阅模型中将回调转换为 Promise

Question

假设我们有一个名为A 的服务，它具有subscribe(callback) 的功能。来自服务的订阅是一个开放的连接，可以随时接收数据，通过回调可以访问数据。 我们可以将此回调转换为承诺吗？如果有，怎么做？

样本

A.subscribe((error, data) => {
    // do something with data
});

Answer 1

我们可以把这个回调转换成 Promise 吗？

不只是一个承诺，不，因为一个承诺只能被解决一次，只有一个履行值（成功时），但你有一个系列值（通常称为“可观察”）。所以你不能将它转换为返回一个承诺，除非你当然希望承诺只用 一个 值（例如第一个）来实现。

您可以将其转换为异步迭代器（可能使用async 生成器函数）。顾名思义，异步迭代器通过返回一个 series 的 Promise 异步提供一系列值。我不能说异步迭代器是否适用于您的用例，但它是最接近我想到的基于 Promise 的可观察对象。

Here's 为 Angular 的 observables 设计的 observable-to-async-iterator 实现，但可以根据需要进行调整。 （遗憾的是，没有指明许可证，所以我无法将其复制到答案中。）

Answer 2

正如 TJ 所建议的，您不能用 single 承诺来表示订阅。我使用了一种类似于 TJ 链接中描述的技术，但在抽象方面有显着差异。下面的 duplexStream 为调用者提供了一个 read 函数，用于您程序中的任何流观察者，以及一个 write 函数以在您的可观察订阅中使用 -

function duplexStream () {
  let t = defer()
  async function* read () {
    while (true) yield await t.deferred
  }
  function write (err, value) {
    if (err) t.reject(err)
    else t.resolve(value)
    t = defer()
  }
  return [read, write]
}

defer 是一个提供外部控制承诺的简单抽象 -

function defer () {
  let resolve, reject
  return { deferred: new Promise((res, rej) => (resolve = res, reject = rej)), resolve, reject }
}

假设我们有两个 HTML 元素 -

<p id="foo"></p>
<p id="bar"></p>

我们现在可以使用 duplexStream 将 observable 转换为异步迭代器。注意read 可以被多次调用，所有读者都将获得传递给write 的值。 write 也可以赋予任意数量的 observables。 duplexStream 是通用的，不绑定到特定的库或类。通过将read 和write 作为通用函数提供给调用者，可以在普通的subscribe 和unsubscribe 事件期间管理这些处理程序-

const [read, write] = duplexStream()

async function update(elem, it) {
  for await (const value of it)
    elem.innerHTML += (value + "<br>")
}

update(document.querySelector("#foo"), read()).catch(console.error)
update(document.querySelector("#bar"), read()).catch(console.error)

A.subscribe(write)

对于这个演示，我们将 A.subscribe 编写为一个模拟 observable，它发出三 (3) 个值，然后是一个错误 -

const A = {
  subscribe (f) {
    setTimeout(f, 1000, null, "one")
    setTimeout(f, 2000, null, "two")
    setTimeout(f, 3000, null, "three")
    setTimeout(f, 4000, Error("something bad"))
  }
}

程序完成后，我们会看到-

<p id="foo">
  one<br>
  two<br>
  three<br>
</p>
<p id="bar">
  one<br>
  two<br>
  three<br>
</p>

并且错误将在console.error 中记录两次，因为每个流阅读器catched -

Error: something bad
Error: something bad

展开下面的sn-p，在自己的浏览器中运行程序验证结果-

function duplexStream () {
  let t = defer()
  async function* read () {
    while (true) yield await t.deferred
  }
  function write (err, value) {
    if (err) t.reject(err)
    else t.resolve(value)
    t = defer()
  }
  return [read, write]
}

function defer () {
  let resolve, reject
  return { deferred: new Promise((res, rej) => (resolve = res, reject = rej)), resolve, reject }
}

const A = {
  subscribe (f) {
    setTimeout(f, 1000, null, "one")
    setTimeout(f, 2000, null, "two")
    setTimeout(f, 3000, null, "three")
    setTimeout(f, 4000, Error("something bad"))
  }
}

const [read, write] = duplexStream()

async function update(elem, it) {
  for await (const value of it)
    elem.innerHTML += (value + "<br>")
}

update(document.querySelector("#foo"), read()).catch(console.error)
update(document.querySelector("#bar"), read()).catch(console.error)

A.subscribe(write)

.as-console-wrapper { max-height: 33% !important; }

<p id="foo"></p>
<p id="bar"></p>