javascript - 在 JavaScript 中实现 Stream

Question

我想实现一个可以执行此操作的流对象:

// a -------1------2----3
// map -----\------\----\
// b --------2------4----6

const a = new Stream();
const b = a.map(value => value * 2);

b.subscribe(console.log);

a.push(1);
// 2
a.push(2);
// 4
a.push(3);
// 6

这里的想法是对象 b 可以为流 a 订阅新的回调。 map 函数应该在 push 被调用时监听并应用映射出的函数以及最初订阅的函数。这是我目前的实现:

class Stream {
  constructor(queue = []) {
    this.queue = queue;
  }

  subscribe(action) {
    if (typeof action === 'function') {
      this.queue.push(action);
    }
  }

  map(callback) {
     this.queue = this.queue.map(
        actionFn => arg => action(callback(arg))
     );

     return this;
  }

  push(value) {
    this.queue.forEach(actionFn => {
      actionFn.call(this, value);
    });
  }
}

当前实现的问题在于，最初 Stream 类中的 queue 是空的，因此它不会通过它。将不胜感激任何建议或帮助。我不想为此使用任何库。

Answer 1

您的 map 需要创建一个新的 Transform 流并返回它。您可以简单地使用标准的 on('data') 事件代替 subscribe ，或者更好地使用 read 方法。

最后 - 您可以简单地使用我的工作，并通过使用 scramjet 有效地实现您的 map 方法，它完全符合您在上面显示的内容 - 而且它支持异步功能。 :)

下面是你如何使用它(在一些 getStream 函数中):

const {DataStream} = require('scramjet');

const stream = new DataStream();

stream.write(1); // you can also use await stream.whenWrote(1);
stream.write(2);
stream.write(3);

return stream.map(x => x * 2);

然后在其他地方阅读:

stream.on('data', x => console.log(`x: ${x}`));
// x: 2
// x: 4
// x: 6

看看 scramjet docs这里