Как транслировать результаты запросов MongoDB с помощью nodejs?

Question

Я искал пример того, как я могу передать результат запроса MongoDB клиенту nodejs.Все решения, которые я нашел до сих пор, похоже, читают результат запроса сразу, а затем отправляют результат обратно на сервер.

Вместо этого я (очевидно) хотел бы предоставить обратный вызов для метода запроса и иметь MongoDBвызовите это, когда следующий кусок набора результатов будет доступен.

Я смотрю на Мангуста - я должен вероятно использовать другой драйвер?

Ян

Answer 1

Потоковое вещание в Mongoose стало доступно в версии 2.4.0, которая появилась через три месяца после того, как вы отправили этот вопрос:

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

Более подробные примеры можно найти на их страница документации .

Answer 2

node-mongodb-driver (нижележащий уровень, который каждый клиент mongoDB использует в nodejs), за исключением API курсора, который упоминали другие, имеет хороший потоковый API ( # 458 ). К сожалению, я не нашел его в другом месте.

Обновление: есть документы также здесь .

Может использоваться следующим образом:

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

Он фактически реализует интерфейс ReadableStream, поэтому у него есть все плюсы (пауза / возобновление и т. Д.)

Answer 3

mongoose на самом деле не "драйвер", это фактически оболочка ORM вокруг драйвера MongoDB (node-mongodb-native).

Чтобы сделать то, что вы делаете, взгляните на метод .find и .each драйвера. Вот код из примеров:

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

Для потоковой передачи результатов вы в основном заменяете sys.puts функцией «потока». Не уверен, как вы планируете передавать результаты. Я думаю, что вы можете сделать response.write() + response.flush(), но вы также можете оформить заказ socket.io.

Answer 4

Вот решение, которое я нашел (пожалуйста, поправьте меня, если это неправильный способ сделать это): (Также извините за плохое кодирование - мне уже слишком поздно это уточнять)

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 5

Я сам изучал потоки mongodb, хотя у меня нет полного ответа, который вы ищете, у меня есть его часть.Вы можете настроить поток socket.io

, используя javascript socket.io и socket.io-streaming, доступные в NPM, также mongodb для базы данных, поскольку использование 40-летней базы данных, в которой есть проблемы, является неправильным, времямодернизировать также 40-летнюю базу данных - это SQL, и, насколько мне известно, SQL не делает потоков

Так что, хотя вы спрашивали только о данных, передаваемых с сервера на клиент, я также хочу получить ответ от клиента к серверу, потому чтоЯ НИКОГДА не могу найти его где-нибудь, когда я ищу, и я хотел настроить одно место с элементами отправки и получения через поток, чтобы каждый мог быстро освоить его.

клиентская сторона отправляет данные на сервер посредством потоковой передачи

stream = ss.createStream();
blobstream=ss.createBlobReadStream(data);
blobstream.pipe(stream);
ss(socket).emit('data.stream',stream,{},function(err,successful_db_insert_id){
 //if you get back the id it went into the db and everything worked
});

сервер получает поток от клиентской стороны и затем отвечает, когда это сделано

ss(socket).on('data.stream.out',function(stream,o,c){
 buffer=[];
 stream.on('data',function(chunk){buffer.push(chunk);});
 stream.on('end',function(){
  buffer=Buffer.concat(buffer);
  db.insert(buffer,function(err,res){
   res=insertedId[0];
   c(null,res);
  });
 });
});

// Это другая половина этой выборки данных и потоковой передачи на клиент

клиентская сторона запрашивает и получает потоковые данные с сервера

stream=ss.createStream();
binarystring='';
stream.on('data',function(chunk){ 
 for(var I=0;i<chunk.length;i++){
  binarystring+=String.fromCharCode(chunk[i]); 
 }
});
stream.on('end',function(){ data=window.btoa(binarystring); c(null,data); });
ss(socket).emit('data.stream.get,stream,o,c);

серверная сторона отвечает на запроспотоковые данные

ss(socket).on('data.stream.get',function(stream,o,c){
 stream.on('end',function(){
  c(null,true);
 });
 db.find().stream().pipe(stream);
});

Самый последний из них - единственный, где я как бы просто вытаскиваю его из задницы, потому что я еще не пробовал, но это должно сработать.На самом деле я делаю что-то подобное, но я записываю файл на жесткий диск, а затем использую fs.createReadStream для потоковой передачи его клиенту.Так что не уверен, что на 100%, но из того, что я прочитал, должно быть, я перезвоню вам, как только протестирую.

Ps, кто-нибудь захочет меня расспросить о моем разговорном способе говорить, я канадеци я люблю говорить "эх", подойди ко мне со своими объятиями и ударишь по братанам / сестренкам: D