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

Question

下面的代码片段是功能性的(从某种意义上说它正在工作;-)),但似乎充其量只是蹩脚......

任何人都可以建议一种方法来使这个更容易组合或至少不那么难看？

该代码基于此页面上的示例: 使用RxJS包装现有API

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);
  }
);

Answer 1

在这个具体的例子中,假设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的文档听起来很复杂,但并不复杂.它只是将源Observable中的事件替换为另一个未来事件.在图表中解释,它x用未来y事件替换每个事件.

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

在我们的例子中,x事件是"成功连接",并且y是"从数据库获得'事物'".

也就是说,根据应用程序的工作方式,有几种不同的方法可以实现.最好将RxJS视为"类固醇上的事件总线"而不是可链接Promise的替代品,因为它实际上不是后者.

如果您将"应用程序中发生的所有事情"建模为事件流,那么在RxJS上进行开发是最好的.如果做得恰当,你不应该看到这些可链接的"做这个,然后那样做,然后做那个",因为最终这是一个必要的范例,而RxJS能够做到这一点.理想情况下,它应该更多地以声明的方式告诉事件是什么.有关更多说明,请参阅本教程,特别是"结束"部分中的讨论.这个要点也许有帮助.