Rx.Observable.forkJoin example

rx-observable-forkjoin.example.js

console.clear();


const perPage = 5;
const total = 23;

function getPage(page){
  var i;
  var data = {
    data: [],
    meta: {
      total: total,
      perPage: perPage,
      pageNum: page
    }
  };
  
  var pageStart = (page-1) * perPage;
  var pageEnd = pageStart + perPage;
  
  for (i=pageStart; i<pageEnd && i<total; i++){
    //console.log(`item: ${i} page: ${page} pageStart ${pageStart} pageEnd ${pageEnd}`)
    data.data.push(i);
  }
  
  console.log(`running get page to get page ${page} ...`);
  
  
  return Rx.Observable.just(data);
}

//var o = Rx.Observable.forkJoin(getPage(1), getPage(2));

var o = getPage(1)
  .flatMap(firstPage => {
    var pages = [Rx.Observable.of(firstPage)];
    
    var i;
    var total = firstPage.meta.total;
    var perPage = firstPage.meta.perPage;
    
    var numPages = Math.ceil(total/perPage);
    for (i=2; i<= numPages; i++){      
        pages.push(getPage(i))
    }
    
    // forkJoin will wait for all the independent 
    // observables being run in parallel to finish
    return Rx.Observable.forkJoin(pages);
  })
  .map(data => {
    
    var results = [];
    data.forEach(item => {
      //console.log('item is:', item)
      results.push(...item.data);
    });
      
    return {
      data: results,
      meta: {
        total: data[0].meta.total
      }
    };
  });

o.subscribe(
  d => console.log(d),
  e => console.error(e),
  () => console.log('done')
);