Javascript (ES6) 可迭代流

Question

是否有使用 ES6 生成器使流可迭代的模式？

请参阅下面的“MakeStreamIterable”。

import {createReadStream} from 'fs'

let fileName = 'largeFile.txt'
let readStream = createReadStream(fileName, {
  encoding: 'utf8',
  bufferSize: 1024
})
let myIterableAsyncStream = MakeStreamIterable(readStream)

for (let data of myIterableAsyncStream) {
  let str = data.toString('utf8')
  console.log(str)
}

我对co 或 bluebird 的coroutine 或使用deasync 阻塞不感兴趣。

金子是 MakeStreamIterable 应该是一个有效的函数。

Answer 1

Is there a pattern for making a stream iterable using ES6 generators?

不，这无法实现，因为生成器是同步的。他们必须知道他们在什么时候产出什么。异步数据源的迭代目前只能通过使用某种基于回调的实现来实现。因此，如果您的意思是“一个有效函数，其结果可以提供给 for-of 循环，则无法使 MakeStreamIterable 成为“有效函数” '.

流是异步的

流表示在可能无限的时间内异步接收的可能无限量的数据。如果我们看一下 definition of an iterator on MDN我们可以更详细地定义什么是使流“可统一”的流:

An object is an iterator when it knows how to access items from a collection one at a time, while keeping track of its current position within that sequence. In JavaScript an iterator is an object that provides a next() method which returns the next item in the sequence. This method returns an object with two properties: done and value.

(重点是我自己。)

让我们从这个定义中挑选出可迭代对象的属性。该对象必须...

1. 知道如何一次访问一个集合中的项目；
2. 能够跟踪其在数据序列中的当前位置；
3. 并提供一个方法 next，该方法检索一个对象，该对象的属性包含序列中的下一个值，或者通知迭代完成.

流不符合上述任何条件，因为...

1. 它无法控制何时它接收数据并且无法“展望 future ”以找到下一个值；
2. 它无法知道何时或是否已收到所有数据，只有在流关闭时才知道；
3. 并且它没有实现 iterable protocol因此不会公开 for-of 可以使用的 next 方法。

______

伪造它(eration)

我们不能实际上迭代从流中接收到的数据(绝对不是使用for-of)，但是我们可以构建一个假装的接口(interface) 通过使用 Promises(耶!)并在闭包中抽象出流的事件处理程序。

// MakeStreamIterable.js
export default function MakeStreamIterable (stream) {
  let collection = []
  let index = 0
  let callback
  let resolve, reject

  stream
    .on('error', err => reject && reject(err))
    .on('end', () => resolve && resolve(collection))
    .on('data', data => {
      collection.push(data)

      try {
        callback && callback(data, index++)
      } catch (err) {
        this.end()
        reject(err)
      }
    })

  function each (cb) {
    if(callback) {
      return promise
    }

    callback = (typeof cb === 'function') ? cb : null

    if (callback && !!collection) {
        collection.forEach(callback)
        index = collection.length
    }

    return promise
  }

  promise = new Promise((res, rej) => {
    resolve = res
    reject = rej
  })

  promise.each = each

  return promise
}

我们可以这样使用它:

import {MakeStreamIterable} from './MakeStreamIterable'

let myIterableAsyncStream = MakeStreamIterable(readStream)

myIterableAsyncStream
  .each((data, i) => {
    let str = data.toString('utf8')
    console.log(i, str)
  })
  .then(() => console.log('completed'))
  .catch(err => console.log(err))

关于此实现的注意事项:

• 不必立即在“可迭代流”上调用each。
• 当 each 被调用时，在其调用之前接收到的所有值都被一个接一个地传递给回调 forEach 风格。之后，所有后续数据都会立即传递给回调。
• 该函数返回一个 Promise，它在流结束时解析完整的数据集合，这意味着我们实际上根本不需要调用 each each 提供的迭代并不令人满意。
• 我培养了将其称为迭代器的错误语义，因此我是一个糟糕的人。请向有关当局举报我。