Abstract query execution. Scroll/scan example

Author: hmalphettes

README.md

@@ -5,8 +5,10 @@ hits and documents responses.
 
 Use case: pipe to and from levelup, pouchdb and other friends.
 
-The client that executes the requests is more or less abstracted at the moment.
-Examples and tests are using the official client.
+The client that executes the requests is wrapped in a closure.
+It is expected to provide the Elasticsearch reponse's body as a JSON.
+
+See the examples and tests with the official Elasticsearch-js client.
 
 # Examples:
 
@@ -33,17 +35,18 @@ require('random-document-stream')(42).pipe(ws);
 var ReadableSearch = require('elasticsearch-streams').ReadableSearch;
 var client = new require('elasticsearch').Client();
 
-var search = {
-  index: 'myindex',
-  from: 0,
-  size: 12,
-  body: {
-    query: { match_all: {} }
-  }
+var searchExec = function searchExec(from, callback) {
+  client.search({
+    index: 'myindex',
+    from: from,
+    size: 12,
+    body: {
+      query: { match_all: {} }
+    }
+  }, callback);
 };
-var queryExec = client.search.bind(client);
 
-var rs = new ReadableSearch(queryExec, search);
+var rs = new ReadableSearch(searchExec);
 var ws = new require('stream').Writable({objectMode:true});
 ws._write = function(chunk, enc, next) {
   console.log('a hit', hit);
@@ -53,15 +56,45 @@ ws._write = function(chunk, enc, next) {
 rs.pipe(ws);
 ```
 
+## Stream scroll/scan results from Elasticsearch
+```
+var scrollExec = function scrollExec(from, callback) {
+  if (this.scroll_id) {
+    return client.scroll({
+      scrollId : this.scroll_id,
+      scroll   : '30s'
+    }, callback);
+  }
+  // get a scroll id first
+  var self = this;
+  client.search({
+    index: 'myindex',
+    scroll: '20s',
+    size: '3',
+    body: {
+      query: { match_all: {} }
+    }
+  }, function(e, resp) {
+    self.scroll_id = resp._scroll_id;
+    callback(e, resp);
+  });
+};
+rs = new ReadableSearch(scrollExec);
+```
+
+## Stream IDs into Elasticsearch multi-get and get documents out.
+```
+# TODO a duplex stream
+```
+
 ## TODO
 ### Short term
 * Document more
 * Multi-get as a duplex
-* Bulk document errors emitted as errors.
-* Search stream: access to total, aggregations etc
+* Bulk document errors should be emitted as errors
 
 ## Later
-Streaming http client or transport
+Streaming http client or elasticsearch-js streaming transport.
 
 # LICENSE
 elasticsearch-streams is freely distributable under the terms of the MIT license.

lib/readable-search.js

@@ -11,18 +11,14 @@ var Readable = require('stream').Readable;
 module.exports = ReadableHits;
 
 /**
- * @param queryExec an executable query functions that takes 2 arguments: the query and its callback.
- * @param pushKeysFirst true to stream the name of the properties first and values second
- *        by default false
+ * @param queryExec an executable query functions that takes 2 arguments: the offset and a callback.
+ * @param limit optional number of hits to stream
  */
-function ReadableHits(queryExec, query, limit, emitSourceOnly) {
+function ReadableHits(queryExec, limit) {
   Readable.call(this, {objectMode:true});
   this.queryExec = queryExec;
-  this.query = query;
-  this.pageSize = query.size || 256;
-  this.emitSourceOnly = !!emitSourceOnly;
   this.total = -1;
-  this.from = query.form || 0;
+  this.from = 0;
   this._next = true;
   if (limit > 0) {
     this.limit = limit;
@@ -48,18 +44,30 @@ ReadableHits.prototype._read = function() {//size) {
 };
 
 ReadableHits.prototype._fetchNextPage = function() {//size) {
-  this.from += this.pageSize;
   var self = this;
-  this.queryExec(this.query, function(e, resp) {
+  this.queryExec(this.from, function(e, resp) {
     self._current = 0;
-    self._hits = resp.hits ? resp.hits.hits : resp.docs.docs;
-    if (self.from + self.pageSize > self._hits.length) {
+    if (e) {
+      self.hits = [];
       self._next = false;
+      self.emit('error', e);
+      return this.push(null);
+    }
+    self.total = resp.hits.total;
+    self._hits = resp.hits.hits;
+    self.from += self._hits.length;
+    if (self.from >= self.total) {
+      self._next = false; // we got them all.
+    }
+    if (!self._hits.length) {
+      // nothing here: end the stream.
+      // self._next = false; // precaution but we should not really need this line.
+      return self.push(null);
     }
     self._shift();
   });
 };
 
-ReadableHits.prototype._shift = function() {//size) {
+ReadableHits.prototype._shift = function() {
   this.push(this._hits[this._current]);
 };

lib/writable-bulk.js

@@ -8,16 +8,23 @@ var Writable = require('stream').Writable;
 module.exports = WritableBulk;
 
 /**
- * @param bulkExec closure pass the bulk cmds as an array and a callback
+ * @param bulkExec closure invoked with the bulk cmds as an array and a callback
  * @param defaults: { op: 'index or create, index by default'
-                      id: 'name of the property that is the id, by default _id',
+ *                    id: 'name of the property that is the id, by default _id',
  *                    _index: 'name of the index or nothing',
  *                    _type:  'name of the type or nothing' }
  *                nothing to emit an error on unknown document command
+ * @param bulkSize number of bulk commands executed at once. 128 by default.
  */
 function WritableBulk(bulkExec, defaults, bulkSize) {
   Writable.call(this, {objectMode:true});
   this.bulkExec = bulkExec;
+
+  if (!bulkSize && typeof defaults === 'number') {
+    bulkSize = defaults;
+    defaults = undefined;
+  }
+
   this.bulkSize = bulkSize || 128;
   this.defaults = defaults || {};
   this.defaults.id = this.defaults.id || '_id';
@@ -56,7 +63,9 @@ WritableBulk.prototype._write = function(chunk, enc, next) {
     }
     if (!this.expectingPayload) {
       if (!chunk.hasOwnProperty('delete')) {
-        this.emit('error', new Error('Unexpected chunk, not index create update delete'));
+        this.emit('error', new Error('Unexpected chunk, not an ' +
+            'index/create/update/delete command and ' +
+            'not a document to index either'));
         return next();
       }
       this.bulkCount++;

package.json

@@ -1,7 +1,7 @@
 {
   "name": "elasticsearch-streams",
   "description": "Stream in and out of Elasticsearch",
-  "version": "0.0.1",
+  "version": "0.0.2",
   "repository": {
     "type": "git",
     "url": "https://github.com/hmalphettes/elasticsearch-streamer.git"

test/test-search.js

@@ -3,57 +3,105 @@ var expect = require('chai').expect;
 var ReadableSearch = require('..').ReadableSearch;
 var Writable = require('stream').Writable;
 var client = new require('elasticsearch').Client();
+var random = require('random-document-stream');
 
 describe('When searching', function() {
   var rs;
   before(function(done) {
-    client.bulk({
-      body: [
-        { index:  { _index: 'myindex', _type: 'mytype', _id: 1 } },
-        { title: 'foo' },
-        { index: { _index: 'myindex', _type: 'mytype', _id: 2 } },
-        { title: 'bar' },
-        { index: { _index: 'myindex', _type: 'mytype', _id: 3 } },
-        { title: 'joe' }
-      ]
-    }, function(e) {
+    populateIndex(42, function(e) {
       if (e) { return done(e); }
-      var params = {
-        index: 'myindex',
-        from: 0,
-        size: 12,
-        body: {
-          query: { match_all: {} }
-        }
+      var queryExec = function(from, callback) {
+        client.search({
+          index: 'myindex',
+          from: from,
+          size: 2,
+          body: {
+            query: { match_all: {} }
+          }
+        }, callback);
       };
-      var queryExec = function(params, callback) {
-        client.search(params, callback);
-      };
-      rs = new ReadableSearch(queryExec, params);
+      rs = new ReadableSearch(queryExec);
       done();
     });
   });
-  it('Must find 3 records', function(done) {
-    var hits = [];
-    var err;
-    var ws = new Writable({objectMode:true});
-    ws._write = function(chunk, enc, next) {
-      hits.push(chunk);
-      expect(chunk._index).to.equal('myindex');
-      expect(chunk._type).to.equal('mytype');
-      expect(chunk._score).to.equal(1);
-      expect(chunk._id).to.exist;
-      expect(chunk._source.title).to.exist;
-      next();
-    };
-    rs.on('error', function(e) {
-      err = e;
-    });
-    rs.on('end', function() {
-      if (err) { return done(err); }
-      expect(hits.length).to.equal(3);
+  it('Must find 42 records by searching them', function(done) {
+    checkRecords(rs, 42, done);
+  });
+});
+
+describe('When scrolling', function() {
+  var rs;
+  before(function(done) {
+    populateIndex(42, function(e) {
+      if (e) { return done(e); }
+      var queryExec = function queryExec(from, callback) {
+        if (this.scroll_id) {
+          return client.scroll({
+            scrollId : this.scroll_id,
+            scroll   : '30s'
+          }, callback);
+        }
+        // get a scroll id first
+        var self = this;
+        client.search({
+          index: 'myindex',
+          scroll: '20s',
+          size: '3',
+          body: {
+            query: { match_all: {} }
+          }
+        }, function(e, resp) {
+          self.scroll_id = resp._scroll_id;
+          callback(e, resp);
+        });
+      };
+      rs = new ReadableSearch(queryExec);
       done();
     });
-    rs.pipe(ws);
   });
-});
+  it('Must find the 42 records by scroll', function(done) {
+    checkRecords(rs, 42, done);
+  });
+});
+
+function populateIndex(nb, done) {
+  client.indices.delete({index:'myindex'}, function() {
+    var cmds = [];
+    var generator = random(0);
+    for (var i = 0; i < nb; i++) {
+      var rec = generator.makeJunk();
+      cmds.push({ index:  { _index: 'myindex', _type: 'mytype', _id: rec._id } });
+      cmds.push(rec);
+    }
+    client.bulk({
+      body: cmds
+    }, function(e) {
+      if (e) { return done(e); }
+      client.indices.refresh({index: 'myindex'}, done);
+    });
+  });
+}
+
+function checkRecords(rs, nb, done) {
+  var hits = [];
+  var err;
+  var ws = new Writable({objectMode:true});
+  ws._write = function(chunk, enc, next) {
+    hits.push(chunk);
+    expect(chunk._index).to.equal('myindex');
+    expect(chunk._type).to.equal('mytype');
+    expect(chunk._score).to.equal(1);
+    expect(chunk._id).to.exist;
+    expect(chunk._source.name).to.exist;
+    next();
+  };
+  rs.on('error', function(e) {
+    err = e;
+  });
+  rs.on('end', function() {
+    if (err) { return done(err); }
+    expect(hits.length).to.equal(nb);
+    done();
+  });
+  rs.pipe(ws);
+}

test/test-write.js

@@ -9,8 +9,8 @@ describe('When writing', function() {
   before(function(done) {
     var bulkExec = function(bulkCmds, callback) {
       client.bulk({
-        index : 'myindex2',
-        type  : 'mytype2',
+        index : 'myindex',
+        type  : 'mytype',
         body  : bulkCmds
       }, callback);
     };
@@ -33,8 +33,8 @@ describe('When writing', function() {
   it('Must have indexed 42 docs', function(done) {
     client.indices.refresh({ index: 'myindex2' }, function() {
       client.count({
-        index: 'myindex2',
-        type: 'mytype2'
+        index: 'myindex',
+        type: 'mytype'
       }, function(e, res) {
         if (e) { return done(e); }
         expect(res.count).to.equal(42);