javascript - 高地在完全消耗之前添加到流中？

Question

是否有一种方法可以向高地流添加一条包含我们需求的完整描述的结束消息？

假设我们有以下情况:

const stream = high([1,4,6,7])

然后通过这个流，我想计算正在处理的每个值并说

sink.drain(stream.pipe(4))

数组元素的数量为 4。考虑到流中可能有数千个对象，我需要从流中消费才能进行计数。

我不能说 array.length 因为它是一个可能带有任何信息的源，并且该信息正在通过流进行处理...我如何向流中添加一个包含所消耗内容的描述的消息结束？

Answer 1

听起来你想围绕你的值(value)观建立某种状态，但又不妨碍你的值(value)观的消耗。我建议查看一些涉及 h.through 的解决方案。

您可以使用 fork 或 observe 拆分流，并从值中减少某些状态:

h([1, 2, 3, 4])
  .through(stream => {
    const length = stream.observe()
      .reduce(0, m => m + 1)
      .map(length => ({ length }))

    return h([stream, length])
      .sequence()
  })
  .errors(err => console.error(err))
  .each(x => console.log(x))

您可以创建一些状态并根据源流的事件返回一个新流:

const h = require('highland')
h([1, 2, 3, 4])
  .through(stream => {
    let length = 0

    return h(push => stream
      .tap(() => ++length)
      .errors(err => push(err))
      .each(x => push(null, x))
      .done(() => {
        push(null, `length: ${length}`)
        push(null, h.nil)
      }))
  })
  .errors(err => console.error(err))
  .each(x => console.log(x))