mongodb - 将两个回调折叠到一个Observable中

下面的代码片段是功能性的（从工作的意义上来说；-），但充其量看起来是蹩脚的……
有谁能建议一种方法来让这件事变得更冷静或者至少不那么难看吗？
代码基于本页的示例：
Wrap an Existing API with RxJS

function connect() {
  return rx.Observable.create(function (observer) {
    mongo.connect('mongodb://127.0.1:27017/things', function(err, db) {
      if(err) observer.onError(err);
      observer.onNext(db);
    });
  }).publish().refCount();
}

function getThings(db) {
  return rx.Observable.create(function (observer) {
    db.collection('things').find().toArray(function(err, results) {
      if(err) observer.onError(err);
      observer.onNext(results);
      observer.onCompleted();
    });
    return function () {
      db.close();
    };
  }).publish().refCount();
}


connect().subscribe(
  function (db) {
    getThings(db).subscribe(console.log);
  }, function (err) {
    console.log(err);
  }
);

最佳答案

在这个具体的例子中，假设getThings()只在connect()发生后发生一次，我将更改getThings()的实现，如下所示：

function getThings() {
  return connect()
    .flatMap(function(db) {
      return rx.Observable.create(function (observer) {
        db.collection('things').find().toArray(function(err, results) {
          if(err) observer.onError(err);
          observer.onNext(results);
          observer.onCompleted();
        });
        return function () {
          db.close();
        };
      });
    });
}

然后您可以订阅getThings()流：

getThings().subscribe(console.log);

我们使用flatMap将连接步骤隐藏在整个getThings()中。flatmap的文档听起来很复杂，但并不复杂。它只是用另一个未来的事件来代替一个来自可观测源的事件。如图所示，它将每个x事件替换为未来的y事件。

---x----------x------->
flatMap( x => --y--> )
------y----------y---->

在我们的例子中，x事件是“成功连接”的，y是“从数据库中获取”的“东西”。
也就是说，有两种不同的方法可以做到这一点，这取决于应用程序应该如何工作。最好将rxjs看作“类固醇事件总线”，而不是可链接承诺的替代品，因为它实际上不是后者。
如果您将“应用程序中发生的一切”建模为事件流，那么在rxjs上开发是最好的。如果做得好，你不应该看到这些可链接的“做这个，然后做那个，然后做那个”，因为最终这是一个命令范型，而rxjs的能力不止于此。理想情况下，它应该更多地以声明的方式讲述事件是什么。请参阅this tutorial了解更多解释，特别是“总结”部分中的论述。这也可能有帮助。