Streams with Promise Transformations in Strategies By providing a stream to each strategy, the strategy can define how it will handle the stream. The foo strategy, for example, batches the items in batches of 10, performs a async operation on the batch that takes 50 milliseconds, and then maps the results. In the bar strategy the stream is simply mapped over one item at a time.
Generators.observe((change) => { let timer = 0 const MILLISECOND_TIMEOUT = 10; const intervalId = setInterval(() => { timer = timer + MILLISECOND_TIMEOUT }, MILLISECOND_TIMEOUT); stream .observe() .map(message => ({ message, elapsedTime: `${timer} milliseconds` })) .pipe(highland().each(change)) .done(() => clearInterval(intervalId) && change('done!')) return () => stream.destroy() && clearInterval(intervalId) && change('done!') })
mutable currentStrategy = 'foo'
You will notice that when processing the stream with the foo strategy, currentNumber will increment by 10 not 1. This happens because foo pulls 10 items off off the stream at a time. Also currentNumber won't change until foo completes processing of the items and requests a new batch.
mutable currentNumber = 1
viewof strategyToggle = { const nextStrategy = currentStrategy === 'foo' ? 'bar' : 'foo'; const toggle = button({title: `Try strategy`, value: nextStrategy}); toggle.addEventListener('click', (event) => { mutable currentStrategy = event.target.value; }); return toggle; }
stream = highland(eternalIterable()) .through(strategies[currentStrategy]) .flatMap(v => highland(formatPromise(v)))
strategies = ({ foo: stream => stream .batch(10) .tap(items => { mutable batches = items }) // this just updates batches for display .flatMap(items => highland(bluebird.delay(50).return(items).map(item => `foo${item}`))) .flatten() , bar: stream => stream.map(item => (`bar${item}`)), })
mutable batches = []
viewof delayTime = { const div = html`<label> Set formatPromise Delay <input type=range min=100 max=2000 step=100> <input type=number min=100 max=2000 step=100 style="width:auto;"> </label> `; const range = div.querySelector("[type=range]"); const number = div.querySelector("[type=number]"); div.value = range.value = number.value = 500; range.oninput = () => number.value = div.value = range.valueAsNumber; number.oninput = () => range.value = div.value = number.valueAsNumber; return div; }
formatPromise = v => bluebird.delay(delayTime).return(`Current value is ${v}`)
{ stream.pipe(highland().each(v => v)) return 'processing stream' }
function* eternalIterable() { while(true) { yield mutable currentNumber++ } }
moment = require('moment')
highland = require('highland@2.13.4/dist/highland.js')
bluebird = require('bluebird')
import {button} from "@jashkenas/inputs"