stress test pipe rxjs

stress-test-rx.ls

{Observable, AsyncSubject} = require \Rx
request = require \request
{sum} = require \prelude-ls

log-source = (source) ->
    source.subscribe do
        (x) ->
            console.log \x, x
        (err) ->
            console.log \err, err
        ->
            console.log \done

epoch = new Date "2015-09-17Z"
date-from = epoch
date-to = new Date (epoch.value-of! + 6 * 3600 * 1000)

to-interval = (start, end, interval) --> 
    [(new Date (start.value-of! + interval) .toJSON!), (new Date (end.value-of! + interval) .toJSON!)]


load-source = (branch-id, [date-from, date-to]) ->
    subject = new AsyncSubject!

    do ->
        err, r, body <- request.get "http://192.168.1.42:4081/apis/branches/#{branch-id}/execute/1000/transformation?campaignId=16944&dateFrom=#{date-from}&dateTo=#{date-to}"
        subject.on-next {date-from, date-to, body: JSON.parse body}
        subject.on-completed!


    subject.asObservable!


load-sources = ([date-from, date-to]:range) ->
    Observable.from <[poJiT0U porWwCl poqBANR]>
    .concatMap (branch-id) -> load-source branch-id, range




source = Observable.interval 250 .time-interval! .map (-> 25 * 1000) .scan do 
    (a, b) -> a + b 
    0
.map (to-interval date-from, date-to)
.concatMap load-sources

log-source source