javascript - 从 ORM 返回的一系列 Promise 生成一个 RxJS Observable

Question

我试图从 SQL 数据库中的一系列(批处理)记录生成一个Observable，我试图遍历数据库中的所有记录。我在 node-js 上使用 ORM，Sequelize 返回包装在 promise 中的记录。

我定义了一个函数 fetchbatch()，它获取下一批并返回一个 Promise[Array[Record]] 和 flatMap'将结果发送到 Observable。

我的条件(终止)在 promise 的 then block 中根据查询是否未返回任何记录设置为全局条件，但永远不会调用回调，只会无限返回 promise，因此永远不会满足终止条件.关于如何处理的任何建议？这是代码的要点。

 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>