JavaScript: сворачивание бесконечных потоков (функция генератора)

Question

В Java можно объявлять и сворачивать бесконечные потоки как так

List<Integer> collect = Stream.iterate(0, i -> i + 2)
    .map(i -> i * 3)
    .filter(i -> i % 2 == 0)
    .limit(10)
    .collect(Collectors.toList());

// -> [0, 6, 12, 18, 24]

В JavaScript я мог использовать функции генератора для получения и распространения потока значений.

// Limit the value in generator
let generator = (function* () {
    for (let i=0; i<10; i++) {
        yield i
    }
})()

[ ...generator ]
    .map(i => i * 3)
    .filter(i => i % 2 === 0)

// -> [0, 6, 12, 18, 24]

Но как я мог согнать и сложить бесконечный поток? Я знаю, что мог бы повторить и ограничить поток с помощью цикла for (n of generator). Но возможно ли это с помощью свободно распространяемого API, такого как пример Java?

Answer 1

Вот альтернативный подход к данному ответу.

1.Функциональный API

Сначала создайте функциональный API.

const itFilter = p => function* (ix) {
  for (const x of ix)
    if (p(x))
      yield x;
};

const itMap = f => function* (ix) {
  for (const x of ix)
    yield f(x);
};

const itTake = n => function* (ix) {
  let m = n;
  
  for (const x of ix) {
    if (m-- === 0)
      break;

    yield x;
  }
};

const comp3 = f => g => h => x =>
  f(g(h(x)));    const xs = [1,2,3,4,5,6,7,8,9,10];

const stream = comp3(itTake(3))
  (itFilter(x => x % 2 === 0))
    (itMap(x => x * 3));

console.log(
  Array.from(stream(xs))
);

2.Box-Type

Затем определите тип Box, чтобы разрешить цепочку методов для произвольно функциональных API.

function Box(x) {
  return new.target ? (this.x = x, this) : new Box(x)
}

Box.prototype.map = function map(f) {return new Box(f(this.x))};
Box.prototype.fold = function fold(f) {return f(this.x)};

3.Цепочка методов

Наконец, используйте новый тип Box для цепочки методов.

const itFilter = p => function* (ix) {
  for (const x of ix)
    if (p(x))
      yield x;
};

const itMap = f => function* (ix) {
  for (const x of ix)
    yield f(x);
};

const itTake = n => function* (ix) {
  let m = n;
  
  for (const x of ix) {
    if (m-- === 0)
      break;
      
    yield x;
  }
};

const xs = [1,2,3,4,5,6,7,8,9,10];

function Box(x) {
  return new.target ? (this.x = x, this) : new Box(x)
}

Box.prototype.map = function map(f) {return new Box(f(this.x))};
Box.prototype.fold = function fold(f) {return f(this.x)};

const stream = Box(xs)
  .map(itMap(x => x * 3))
  .map(itFilter(x => x % 2 === 0))
  .map(itTake(3))
  .fold(x => x);
  
 console.log(
   Array.from(stream)
 );

Box дает вам свободный API бесплатно.

Answer 2

Я добавлю еще один ответ, который может быть тем, что вы ищете. Я являюсь автором scramjet платформы, основанной на потоках, которая добавляет плавный API к преобразованиям. То, что вы хотели, может быть достигнуто довольно легко с ним:

import {DataStream} from "scramjet";
let i = 0;
const out = await (
    DataStream.from(function*() { let n = 2; while (true) yield n++; })
        .map(n => n+2)
        .filter(i -> i % 2 == 0)
        .until(() => i++ === 10)
        .toArray()
);

Я построил его в основном для асинхронных операций (так что вы можете просто заменить любую из этих функций на асинхронные, и она будет работать точно так же). Так что ответ, если это возможно, - да.

Одно замечание: потоки node.js, на которых это основано, содержат в себе несколько буферов, поэтому генератор, вероятно, будет повторяться пару раз больше, чем позволяет метод before.

Answer 3

Вот пример -

// a terminating generator
const range = function* (from, to)
{ while (from < to)
    yield from++
}

// higher-order generator
const G =
  range(0, 100).filter(isEven).map(square)

for (const x of G)
  console.log(x)

// (0*0) (2*2) (4*4) (6*6) (8*8) ...
// 0 4 16 36 64 ...

Мы можем сделать что-то подобное, расширив прототип генератора -

const Generator =
  Object.getPrototypeOf(function* () {})

Generator.prototype.map = function* (f, context)
{ for (const x of this)
    yield f.call(context, x)
}

Generator.prototype.filter = function* (f, context)
{ for (const x of this)
    if (f.call(context, x))
      yield x
}

Разверните фрагмент ниже, чтобы проверить наш прогресс в вашем браузере -

const Generator =
  Object.getPrototypeOf(function* () {})

Generator.prototype.map = function* (f, context)
{ for (const x of this)
    yield f.call(context, x)
}

Generator.prototype.filter = function* (f, context)
{ for (const x of this)
    if (f.call(context, x))
      yield x
}

// example functions
const square = x =>
  x * x

const isEven = x =>
  (x & 1) === 0
  
// an terminating generator
const range = function* (from, to)
{ while (from < to)
    yield from++
}

// higher-order generator
for (const x of range(0, 100).filter(isEven).map(square))
  console.log(x)

// (0*0) (2*2) (4*4) (6*6) (8*8) ...
// 0 4 16 36 64 ...

Продолжая, что-то вроде fold или collect предполагает, что поток в конечном итоге завершается, в противном случае он не может вернуть значение -

Generator.prototype.fold = function (f, acc, context)
{ for (const x of this)
    acc = f.call(context, acc, x)
  return acc
}

const result =
  range(0, 100)      // <- a terminating stream
    .filter(isEven)
    .map(square)
    .fold(add, 0)    // <- assumes the generator terminates

console.log(result)
// 161700

Если вам нужно сложить бесконечный поток, вы можете реализовать limit -

Generator.prototype.limit = function* (n)
{ for (const x of this)
    if (n-- === 0)
      break // <-- stop the stream
    else
      yield x
}

// an infinite generator
const range = function* (x = 0)
{ while (true)
    yield x++
}

// fold an infinite stream using limit
const result =
  range(0)          // infinite stream, starting at 0
    .limit(100)     // limited to 100 values
    .filter(isEven) // only pass even values
    .map(square)    // square each value
    .fold(add, 0)   // fold values together using add, starting at 0

console.log(result)
// 161700

Разверните фрагмент кода ниже, чтобы проверить результат в своем браузере -

const Generator =
  Object.getPrototypeOf(function* () {})

Generator.prototype.map = function* (f, context)
{ for (const x of this)
    yield f.call(context, x)
}

Generator.prototype.filter = function* (f, context)
{ for (const x of this)
    if (f.call(context, x))
      yield x
}

Generator.prototype.fold = function (f, acc, context)
{ for (const x of this)
    acc = f.call(context, acc, x)
  return acc
}

Generator.prototype.limit = function* (n)
{ for (const x of this)
    if (n-- === 0)
      break // <-- stop the stream
    else
      yield x
}

const square = x =>
  x * x

const isEven = x =>
  (x & 1) === 0
  
const add = (x, y) =>
  x + y

// an infinite generator
const range = function* (x = 0)
{ while (true)
    yield x++
}

// fold an infinite stream using limit
const result =
  range(0)          // starting at 0
    .limit(100)     // limited to 100 values
    .filter(isEven) // only pass even values
    .map(square)    // square each value
    .fold(add, 0)   // fold values together using add, starting at 0

console.log(result)
// 161700

Выше обратите внимание, как изменение порядка limit на после выражения filter меняет результат -

const result =
  range(0)          // starting at 0
    .filter(isEven) // only pass even values
    .limit(100)     // limited to 100 values
    .map(square)    // square each value
    .fold(add, 0)   // fold values together using add, starting at 0

console.log(result)
// 1313400

В первой программе -

1. начать с бесконечного диапазона (0, 1, 2, 3, 4, ...)
2. ограничение до 100 значений (0, 1, 2, 3, 4, ...,97, 98, 99)
3. только передать четные значения (0, 2, 4, ...94, 96, 98)
4. квадрат каждого значения (0, 4, 16, ..., 8836, 9216, 9604)
5. сворачивать значения, используя add, начиная с 0, (0 + 0 + 4 + 16 + ..., + 8836 + 9216 + 9604)
6. результат 161700

Во второй программе -

1. начать с бесконечного диапазона (0, 1, 2, 3, 4, ...)
2. только передать четные значения (0, 2, 4, ...)
3. ограничение до 100 значений (0, 2, 4, 6, 8, ...194, 196, 198)
4. квадрат каждого значения (0, 4, 16, 36, 64, ..., 37636, 38416, 29304)
5. сложите значения, используя add, начиная с 0, (0 + 4 + 16 + 36 + 64 + ..., + 37636+ 38416 + 29304)
6. результат 1313400

---

Наконец, мы реализуем collect, который, в отличие от fold, не запрашивает начальный аккумулятор. Вместо этого первое значение вручную выкачивается из потока и используется в качестве начального аккумулятора. Поток возобновляется, складывая каждое значение с предыдущим -

Generator.prototype.collect = function (f, context)
{ let { value } = this.next()
  for (const x of this)
    value = f.call(context, value, x)
  return value
}

const toList = (a, b) =>
  [].concat(a, b)

range(0,100).map(square).collect(toList)
// [ 0, 1, 2, 3, ..., 97, 98, 99 ]

range(0,100).map(square).collect(add)
// 4950

---

И остерегайтесь двойного потребления ваших потоков! JavaScript не дает нам постоянных итераторов, поэтому после использования потока вы не можете надежно вызывать другие функции более высокого порядка в потоке -

// create a stream
const stream  =
  range(0)
    .limit(100)
    .filter(isEven)
    .map(square)

console.log(stream.fold(add, 0)) // 161700
console.log(stream.fold(add, 0)) // 0 (stream already exhausted!)

// create another stream
const stream2  =
  range(0)
    .limit(100)
    .filter(isEven)
    .map(square)

console.log(stream2.fold(add, 0)) // 161700
console.log(stream2.fold(add, 0)) // 0 (stream2 exhausted!)

Это может произойти, когда вы делаете что-то вроде merge -

const r =
  range (0)

r.merge(r, r).limit(3).fold(append, [])
// double consume! bug!
// [ [ 0, 1, 2 ], [ 3, 4, 5 ], [ 6, 7, 8 ] ]
// expected:
// [ [ 0, 0, 0 ], [ 1, 1, 1 ], [ 2, 2, 2 ] ]

// fresh range(0) each time
range(0).merge(range(0), range(0)).limit(3).fold(append, [])
// correct:
// [ [ 0, 0, 0 ], [ 1, 1, 1 ], [ 2, 2, 2 ] ]

Использование генератора fresh (range(0)...) каждый раз позволяет избежать проблемы -

const stream =
  range(0)
    .merge
      ( range(0).filter(isEven)
      , range(0).filter(x => !isEven(x))
      , range(0).map(square)
      )
    .limit(10)

console.log ('natural + even + odd + squares = ?')
for (const [ a, b, c, d ] of stream)
  console.log (`${ a } + ${ b } + ${ c } + ${ d } = ${ a + b + c + d }`)

// natural + even + odd + squares = ?
// 0 + 0 + 1 + 0 = 1
// 1 + 2 + 3 + 1 = 7
// 2 + 4 + 5 + 4 = 15
// 3 + 6 + 7 + 9 = 25
// 4 + 8 + 9 + 16 = 37
// 5 + 10 + 11 + 25 = 51
// 6 + 12 + 13 + 36 = 67
// 7 + 14 + 15 + 49 = 85
// 8 + 16 + 17 + 64 = 105
// 9 + 18 + 19 + 81 = 127

Это основная причина использования параметров для наших генераторов: это заставит вас задуматься о правильном их повторном использовании. Таким образом, вместо определения stream как const выше, наши потоки должны всегда быть функциями, даже если нулевые -

// streams should be a function, even if they don't accept arguments
// guarantees a fresh iterator each time
const megaStream = (start = 0, limit = 1000) =>
  range(start) // natural numbers
    .merge
      ( range(start).filter(isEven) // evens
      , range(start).filter(x => !isEven(x)) // odds
      , range(start).map(square) // squares
      )
    .limit(limit)

const print = s =>
{ for (const x of s)
    console.log(x)
}

print(megaStream(0).merge(megaStream(10, 3)))
// [ [ 0, 0, 1, 0 ], [ 10, 10, 11, 100 ] ]
// [ [ 1, 2, 3, 1 ], [ 11, 12, 13, 121 ] ]
// [ [ 2, 4, 5, 4 ], [ 12, 14, 15, 144 ] ]

print(megaStream(0).merge(megaStream(10), megaStream(100)).limit(5))
// [ [ 0, 0, 1, 0 ], [ 10, 10, 11, 100 ], [ 100, 100, 101, 10000 ] ]
// [ [ 1, 2, 3, 1 ], [ 11, 12, 13, 121 ], [ 101, 102, 103, 10201 ] ]
// [ [ 2, 4, 5, 4 ], [ 12, 14, 15, 144 ], [ 102, 104, 105, 10404 ] ]
// [ [ 3, 6, 7, 9 ], [ 13, 16, 17, 169 ], [ 103, 106, 107, 10609 ] ]
// [ [ 4, 8, 9, 16 ], [ 14, 18, 19, 196 ], [ 104, 108, 109, 10816 ] ]

Мы можем реализовать merge как -

Generator.prototype.merge = function* (...streams)
{ let river = [ this ].concat(streams).map(s => [ s, s.next() ])
  while (river.every(([ _, { done } ]) => done === false))
  { yield river.map(([ _, { value } ]) => value)
    river = river.map(([ s, _ ]) => [ s, s.next() ])
  }
}

Разверните фрагмент кода ниже, чтобы проверить результат в своем браузере -

const Generator =
  Object.getPrototypeOf(function* () {})

Generator.prototype.map = function* (f, context)
{ for (const x of this)
    yield f.call(context, x)
}

Generator.prototype.filter = function* (f, context)
{ for (const x of this)
    if (f.call(context, x))
      yield x
}

Generator.prototype.limit = function* (n)
{ for (const x of this)
    if (n-- === 0)
      break // <-- stop the stream
    else
      yield x
}

Generator.prototype.merge = function* (...streams)
{ let river = [ this ].concat(streams).map(s => [ s, s.next() ])
  while (river.every(([ _, { done } ]) => done === false))
  { yield river.map(([ _, { value } ]) => value)
    river = river.map(([ s, _ ]) => [ s, s.next() ])
  }
}

const isEven = x =>
  (x & 1) === 0

const square = x =>
  x * x

const range = function* (x = 0)
{ while (true)
    yield x++
}

// streams should be functions, even if they don't have parameters
const megaStream = (start = 0, limit = 1000) =>
  range(start) // natural numbers
    .merge
      ( range(start).filter(isEven) // evens
      , range(start).filter(x => !isEven(x)) // odds
      , range(start).map(square) // squares
      )
    .limit(limit)

// for demo only
const print = s =>
{ for (const x of s) console.log(x) }

print(megaStream(0).merge(megaStream(10, 3)))
// [ [ 0, 0, 1, 0 ], [ 10, 10, 11, 100 ] ]
// [ [ 1, 2, 3, 1 ], [ 11, 12, 13, 121 ] ]
// [ [ 2, 4, 5, 4 ], [ 12, 14, 15, 144 ] ]

print(megaStream(0).merge(megaStream(10), megaStream(100)).limit(5))
// [ [ 0, 0, 1, 0 ], [ 10, 10, 11, 100 ], [ 100, 100, 101, 10000 ] ]
// [ [ 1, 2, 3, 1 ], [ 11, 12, 13, 121 ], [ 101, 102, 103, 10201 ] ]
// [ [ 2, 4, 5, 4 ], [ 12, 14, 15, 144 ], [ 102, 104, 105, 10404 ] ]
// [ [ 3, 6, 7, 9 ], [ 13, 16, 17, 169 ], [ 103, 106, 107, 10609 ] ]
// [ [ 4, 8, 9, 16 ], [ 14, 18, 19, 196 ], [ 104, 108, 109, 10816 ] ]