Jak przesyłać strumieniowo wyniki zapytania MongoDB z nodejs?

Question

Szukałem przykładu, w jaki sposób mogę przesłać wynik zapytania MongoDB do klienta nodejs. Wszystkie znalezione dotąd rozwiązania wydają się od razu odczytywać wynik zapytania, a następnie wysłać wynik z powrotem do serwera.

Zamiast tego (oczywiście) chciałbym dostarczyć wywołanie zwrotne do metody kwerendy i zadzwonić do MongoDB, gdy następna porcja zestawu wyników jest dostępna.

Szukałem mangusty - czy powinienem użyć innego sterownika?

Jan

Answer 1

żywo w Mongoose stał się dostępny w wersji 2.4.0, który ukazał three months po tym, jak pisał to pytanie:

Model.where('created').gte(twoWeeksAgo).stream().pipe(writeStream); 

Więcej opracowane przykłady można znaleźć na ich documentation page.

Answer 2

mongoose nie jest naprawdę "kierowca", to rzeczywiście wrapper ORM wokół kierowcy MongoDB (node-mongodb-native).

Aby zrobić to, co robisz, przyjrzyj się metodzie sterownika .find i .each. Oto niektóre kodu z przykładów:

// Find all records. find() returns a cursor 
collection.find(function(err, cursor) { 
    sys.puts("Printing docs from Cursor Each") 
    cursor.each(function(err, doc) { 
    if(doc != null) sys.puts("Doc from Each " + sys.inspect(doc)); 
    })      
}); 

Aby przesyłać wyniki, jesteś w zasadzie zastępowania że sys.puts ze swojej funkcji „Stream”. Nie wiesz, jak chcesz przesyłać strumieniowo wyniki. Myślę, że możesz zrobić response.write() + response.flush(), ale możesz też sprawdzić płatność socket.io.

Answer 3

Oto rozwiązanie znalazłem (proszę mnie poprawić, ktoś jeśli thatis niewłaściwy sposób to zrobić): (także usprawiedliwiać złe kodowanie - zbyt późno dla mnie teraz, aby upiększać tego)

var sys = require('sys') 
var http = require("http"); 

var Db = require('/usr/local/src/npm/node_modules/mongodb/lib/mongodb').Db, 
    Connection = require('/usr/local/src/npm/node_modules/mongodb/lib/mongodb').Connection, 
    Collection = require('/usr/local/src/npm/node_modules/mongodb/lib/mongodb').Collection, 
    Server = require('/usr/local/src/npm/node_modules/mongodb/lib/mongodb').Server; 

var db = new Db('test', new Server('localhost',Connection.DEFAULT_PORT , {})); 

var products; 

db.open(function (error, client) { 
    if (error) throw error; 
    products = new Collection(client, 'products'); 
}); 

function ProductReader(collection) { 
     this.collection = collection; 
} 

ProductReader.prototype = new process.EventEmitter(); 

ProductReader.prototype.do = function() { 
     var self = this; 

     this.collection.find(function(err, cursor) { 
       if (err) { 
         self.emit('e1'); 
         return; 

       } 
       sys.puts("Printing docs from Cursor Each"); 

       self.emit('start'); 
       cursor.each(function(err, doc) { 
         if (!err) { 
           self.emit('e2'); 
           self.emit('end'); 
           return; 
         } 

         if(doc != null) { 
           sys.puts("doc:" + doc.name); 
           self.emit('doc',doc); 
         } else { 
           self.emit('end'); 
         } 
       }) 
     }); 
}; 
http.createServer(function(req,res){ 
     pr = new ProductReader(products); 
     pr.on('e1',function(){ 
       sys.puts("E1"); 
       res.writeHead(400,{"Content-Type": "text/plain"}); 
       res.write("e1 occurred\n"); 
       res.end(); 
     }); 
     pr.on('e2',function(){ 
       sys.puts("E2"); 
       res.write("ERROR\n"); 
     }); 

     pr.on('start',function(){ 
       sys.puts("START"); 
       res.writeHead(200,{"Content-Type": "text/plain"}); 
       res.write("<products>\n"); 
     }); 

     pr.on('doc',function(doc){ 
       sys.puts("A DOCUMENT" + doc.name); 
       res.write("<product><name>" + doc.name + "</name></product>\n"); 
     }); 

     pr.on('end',function(){ 
       sys.puts("END"); 
       res.write("</products>"); 
       res.end(); 
     }); 

     pr.do(); 

    }).listen(8000); 

Answer 4

node-mongodb-driver (podstawowa warstwa, którą każdy klient mongoDB używa w nodejs) z wyjątkiem API kursora, o którym wspomniano, że ma ładny strumień API (#458). Niestety nie znalazłem tego udokumentowanego gdzie indziej.

Aktualizacja: there are docs również here.

Może być stosowany tak:

var stream = collection.find().stream() 
stream.on('error', function (err) { 
    console.error(err) 
}) 
stream.on('data', function (doc) { 
    console.log(doc) 
}) 

To faktycznie implementuje interfejs ReadableStream, a więc ma wszystkie cukierki (pauza/wznowienie etc)