如何从mongo游标查询返回observable.fromEvent？

Posted 2021-04-18

我有一个函数执行查询并从查询返回的游标事件返回一个observable：

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:
...

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

如您所见，Cursor实际上触发了一个名为data的事件：qazxsw poi

答案

http://mongodb.github.io/node-mongodb-native/3.0/api/Cursor.html#event:data运算符接收可观察量的do，next和error通知，但对可观察量没有影响。也就是说，忽略complete运算符的do函数返回的任何值。因此，传递给next的函数接收subscribe。

您最有可能想使用Db而不是do，将可观察事件展平为可观察流：

switchMap

另一答案

我发现使用以下代码更方便

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

原因是方法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() ) }) } 忽略了光标的“完整”事件，因此您永远不能输入订户的“onComplete”功能。

另一方面，使用Observable.from方法允许控制光标的完成，因此也触发用户的“onComplete”功能。