MongoDB Tailable Курсор в nodejs, как остановить поток

Question

я использую ниже код, чтобы получить данные из MongoDB ограничена коллекцией

функции слушать (условия, обратный вызов) {

 db.openConnectionsNew([req.session.client_config.db] , function(err, conn){ 

      if(err) {console.log({err:err}); return next(err);} 

      coll = db.opened[db_name].collection('messages'); 

     latestCursor = coll.find(conditions).sort({$natural: -1}).limit(1) 
     latestCursor.nextObject(function(err, latest) { 
      if (latest) { 
       conditions._id = {$gt: latest._id} 
      } 
      options = { 
       tailable: true, 
       awaitdata: true, 
       numberOfRetries: -1 
      } 
      stream = coll.find(conditions, options).sort({$natural: -1}).stream() 
      stream.on('data', callback) 
     }); 

     }); 
    } 

, а затем я использую сокеты. вещания (roomName, 'данные', документ);

на стороне клиента

io.socket.get('/get_messages/', function(resp){ 
}); 
io.socket.on('data', function notificationReceivedFromServer (data) { 
    console.log(data); 

});

это прекрасно работает, так как я могу видеть любой новый документ, который вставлен в db.

Я могу видеть в mongod -verbose, что после каждых 200 мс существует запрос, выполняющийся с запросом {$ gt: latest_id}, и это прекрасно, но я понятия не имею, как я могу закрыть этот запрос :(Я очень новый в nodejs и используя MongoDB tailable опции в первый раз, и я полностью потерял, любая помощь или подсказка высоко ценится

Answer 1

Что возвращается из метода .stream() от объекта Cursor вернулся из .find() является реализацией интерфейса node transform stream. В частности, это «читаемый» поток.

Как таковое, это событие «данных» излучается всякий раз, когда появляются новые данные r доступный и доступный в потоке для чтения.

Существуют и другие методы, такие как .pause() и .resume(), которые могут использоваться для управления потоком данных событий. Как правило, вы бы назвали этот «внутри» а «данные» обратным вызов события, где вы хотели, чтобы убедиться, что код в этом функции обратного вызова был выполнен до события «следующих» данные было обработано:

stream.on("data", function(data) { 
    // pause before processing 
    stream.pause(); 

    // do some work, possibly with a callback 
    something(function(err,result) { 

     // Then resume when done 
     stream.resume(); 
    }); 
}); 

Но это, конечно, просто вопрос «обзора». Таким образом, до тех пор, пока переменная «stream» определена в области, где другой кусок кода может получить к ней доступ, вы можете вызывать любой метод в любое время.

Опять же, с той же точки зрения, вы можете просто «определить» объект «поток» в любой точке кода, что делает «обработку событий» избыточной.

// Just overwrite the object 
scope = undefined; 

Стоит знать. Фактически, новая «версия 2.x» драйвера узла обертывает «интерфейс потока» непосредственно в стандартный объект Cursor без необходимости вызывать .stream() для преобразования. Потоки узлов - очень полезные и мощные вещи, которые стоило бы сблизиться с их использованием.