Generate RxJS Observed from the Promises sequence returned from ORM

Question

I am trying generatea Observablefrom a sequence of (packed) records from an SQL database, I am trying to run all the records in a database. I use ORM on node-js, Sequelizewhich returns records wrapped in a promise.

I defined a function fetchbatch()that fetches the next batch and returns Promise[Array[Record]]and flatMap'result in Observable.

My condition (for completion) is set as global in the promise block, based on the fact that the request did not return any records, but the callback is never called, only promises returns endlessly, and therefore the completion condition is never satisfied. Any suggestions on how this is handled? Here is the gist of the code.

 function getAllPaginated(conditions) {
    var remaining = true;
    var batch_size = 20;
    function condition(){ return remaining; }
    function selector(promisedBatchOfRecords){
      //console.log(promisedBatchOfRecords);
      //return Observable.fromPromise(promisedBatchOfRecords[1]);
      return (promisedBatchOfRecords[1]);
    }
    function fetchBatch(batchNumberAndBatch) { // Returns [NextBatchNumber, Promise[Array[Record]]]
      //console.log(remaining);
      var batch_number = batchNumberAndBatch[0];
      var offset = (batch_number - 1) * batch_size;
      var rs = Records.findAll({where: conditions, offset: offset, limit: batch_size});
      return [batch_number + 1,
              rs.then(function(batch) {
                console.log(batch.length);
                if (!(batch.length > 0)){
                  remaining = false;
                };
                return batch.map(function(r){r.dataValues});
              })];
    }
    return Observable.generate(fetchBatch([1, []]), condition, fetchBatch).flatMap(Ramda.identity/*over the promise*/).flatMap(Ramda.identity/*over the list*/);
  }
var o = getAllPaginated({where: {a: "b"}})
o.subScribeOnNext(console.log)

Answer 1

- :

const result = new Rx.Subject;
const batch_size = 3;

// Init the recursion
whileFind(0)
  .subscribe();

// Grab the result here
result
  .mergeAll()
  .map(batch => batch.dataValues)
  .subscribe(value => console.log(value));
    
// Recursion function
function whileFind(offset) {
  return Rx.Observable.fromPromise(findAll(offset))
    .concatMap(batch => {
      if (batch.length <= 0) { // Stop condition
        return Rx.Observable.of(null);
      }
      else {
        result.next(batch); // Push the chunk to the result
        return whileFind(offset + batch_size);
      }
  });
}

// Emulate Records.findAll from your BO
function findAll(offset): Promise<Object[]> {
  const data = [
    { dataValues: 1 },
    { dataValues: 2 },
    { dataValues: 3 },
    { dataValues: 4 },
    { dataValues: 5 },
    { dataValues: 6 },
    { dataValues: 7 },
    { dataValues: 8 },
    { dataValues: 9 },
    { dataValues: 10 }
  ];

  return Promise.resolve(data.slice(offset, offset + batch_size));
}

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.0-beta.12/Rx.min.js"></script>

Hide result