Реализация потока в JavaScript

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

Проблема с текущей реализацией заключается в том, что изначально queue в классе Stream пуст, поэтому он не проходит через него.Буду признателен за любые предложения или помощь.Я не хотел бы использовать для этого какую-либо библиотеку.

Answer 1

Ваша карта должна создать новый поток Transform и вернуть его.Вместо subscribe вы можете просто использовать стандартное событие on('data') или, что лучше, использовать метод read.

Наконец - вы можете просто использовать мою работу, и ваш метод map уже будет эффективно реализованиспользуя scramjet, который делает именно то, что вы показали выше - и к тому же он поддерживает асинхронные функции.:)

Вот как вы можете использовать его (в некоторой функции 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 здесь