如何从 mongo 游标查询中 return observable.fromEvent？

Question

我有一个执行查询的函数和 returns 从游标事件中观察到的查询 returns:

exports.query_tokens = (db) => {
  var req = db.collection('collectionName').find({});
  return Rx.Observable.fromEvent(req, 'data');
}

我是这样使用它的：

...
do(mongo_functions.query_tokens).
subscribe(console.log);

但我在控制台中得到了这个：

Db {
nodejs                    |   domain: null,
nodejs                    |   _events: {},
nodejs                    |   _eventsCount: 0,
nodejs                    |   _maxListeners: undefined,
nodejs                    |   s: 
nodejs                    |    { databaseName: 'myDatabase',
nodejs                    |      dbCache: {},
nodejs                    |      children: [],
nodejs                    |      topology: 
nodejs                    |       Server {
nodejs                    |         domain:
...

如您所见，它们不是我的文件。我做错了什么？

如您所见，Curso 实际上触发了一个名为数据的事件：http://mongodb.github.io/node-mongodb-native/3.0/api/Cursor.html#event:data

Answer 1

do 运算符接收 observable 的 next、error 和 complete 通知，但对 observable 没有影响。也就是说，从 do 运算符的 next 函数返回的任何值都将被忽略。因此，传递给 subscribe 的函数接收 Db.

而不是 do，您很可能想使用 switchMap，将事件可观察到可观察流：

...
.switchMap(mongo_functions.query_tokens)
.subscribe(console.log);

Answer 2

我发现使用下面的代码更方便

export function findObs(collection: Collection<any>, queryConditions?: any) {
    const queryObj = queryConditions ? queryConditions : {};
    const queryCursor = collection.find(queryObj);
    return Observable.create((observer: Observer<Array<ObjectID>>): TeardownLogic => {
                            queryCursor.forEach(
                                doc => observer.next(doc),
                                () => observer.complete()
                            )
                        })
}

原因是方法Observable.from忽略了光标的"complete"事件，所以你永远无法进入 "onComplete"订阅者的功能。

另一方面，使用方法Observable.create允许控制光标的完成，因此也触发订阅者的"onComplete"函数.

如何从 mongo 游标查询中 return observable.fromEvent？

How to return an observable.fromEvent from a mongo cursor query?

mongodb

node.js

rxjs

reactivex