Help to build control flow func

[eric-basley]

I can't find a way to build a control flow system to pop values from a stream until a concurrency limit.

const urls = reduce((stream, url) => stream(url), flyd.stream(), R.times(makeUrl(), 1000)); const results = throughLimit(requestUrl, urls, { maxConcurrency }); // only maxConcurrency requests can be pending at a time // result from requestUrl() are pushed one by one in results

Thanks for your help.

Eric.

[eric-basley]

I've made this implementation, but behavior of count() is very weird !!!

const throughLimit= (fn, stream, maxConcurrency) => { const running = flyd.stream(0); const cache = []; return flyd.combine((s, count, self, changed) => { if(count() < maxConcurrency) { count(count() + 1); const value = includes(s, changed) ? s() : cache.pop(); if(value) fn(value, res => { count(count() - 1); self(res) }); }else if (includes(s, changed)){ cache.unshift(s()); } }, [stream, running]); };

Can we update a dependant stream inside a combine() body ?

[eric-basley]

let's forget the first buggy implementation! here's a first version, with a mix of stream and recursive callbacks.

it's called as: const resStream = throughLimit(makeRequest, ulrsStream, 10)

Does anyone have a better idea ?

const throughLimit= (fn, stream, maxConcurrency) => { let running = 0; const cache = []; const callFn = (value, cb) => { fn(value, res => { cb(res); if(cache.length) { const value = cache.pop(); running++; callFn(value, cb); } }); }; return flyd.combine((s, self, changed) => { if(running < maxConcurrency) { running++; callFn(s(), res => { running--; self(res); }); }else{ cache.unshift(s()); } }, [stream]); };

[nordfjord]

You could use ramda to help you there

Maybe something like this could do the trick for you?

const urlStream = flyd.stream() // Stream string

const joinedUrlStream = urlStream
  .pipe(flyd.scan(R.flip(R.append), [])) // Stream [string]

const requestStream = joinedUrlStream
  .map(R.splitEvery(1000)) // Stream [[string]]
  .map(async splitUrls => {
    let result = []
    for (const urls of splitUrls) {
      result = result.concat(await Promise.all(urls.map(makeRequest)))
    }
    return result
  }) // Stream Promise [Result]

const responseStream = requestStream
  .chain(flyd.fromPromise) // Stream [Result]