Initial commit

.gitignore

@@ -0,0 +1,24 @@
+node_modules
+.DS_Store
+*~
+coverage
+Thumbs.db
+.bak
+.tmp
+
+lib-cov
+*.seed
+*.log
+*.dat
+*.out
+*.pid
+*.gz
+
+pids
+logs
+results
+
+npm-debug.log
+
+test_migration_remote
+pouch__all_dbs__

.jshintrc

@@ -0,0 +1,34 @@
+{
+  "proto": true,
+  "browser": true,
+  "curly": true,
+  "devel": true,
+  "eqeqeq": true,
+  "eqnull": true,
+  "evil": false,
+  "immed": false,
+  "jquery": true,
+  "latedef": false,
+  "laxcomma": true,
+  "newcap": true,
+  "node": true,
+  "noempty": true,
+  "nonew": true,
+  "predef":
+  [
+    "after",
+    "afterEach",
+    "before",
+    "beforeEach",
+    "describe",
+    "it",
+    "unescape",
+    "setImmediate"
+  ],
+  "smarttabs": true,
+  "trailing": false,
+  "undef": true,
+  "unused": true,
+  "strict": false,
+  "expr": true
+}

README.md

@@ -0,0 +1,70 @@
+# Elasticsearch streams
+
+Expose a Writeable stream for bulk commands and a Readable stream from
+hits and documents responses.
+
+Use case: pipe to and from levelup, pouchdb and other friends.
+
+The client is more or less agnostic at the moment.
+
+Examples:
+
+## Stream random records into Elasticsearch
+```
+var client = new require('elasticsearch').Client();
+
+var bulkExec = function(bulkCmds, callback) {
+  client.bulk({
+    index : 'myindex',
+    type  : 'mytype',
+    body  : bulkCmds
+  }, callback);
+};
+var ws = new WritableBulk(bulkExec);
+require('random-document-stream')(42).pipe(ws);
+```
+
+## Stream search results into Elasticsearch
+```
+var ReadableSearch = require('elasticsearch-streams')
+var client = new require('elasticsearch').Client();
+
+var search = {
+  index: 'myindex',
+  from: 0,
+  size: 12,
+  body: {
+    query: { match_all: {} }
+  }
+};
+var queryExec = client.search.bind(client);
+
+var rs = new ReadableSearch(queryExec, search);
+rs.pipe(ws);
+```
+
+## TODO
+### Short term
+* Document more
+* Handle errors correctly
+
+## Later
+Streaming http client
+
+# LICENSE
+elasticsearch-streams is freely distributable under the terms of the MIT license.
+
+Copyright (c) 2014 Sutoiku, Inc.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
+documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
+rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit
+persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
+Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
+WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
+OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

index.js

@@ -0,0 +1,4 @@
+module.exports = {
+  WritableBulk: require('./lib/writable-bulk'),
+  ReadableSearch: require('./lib/readable-search')
+};

lib/readable-hits.js

@@ -0,0 +1,65 @@
+/**
+ * Expose an elasticsearch query that returns hits or docs as a stream of hits or docs.
+ *
+ * Expect the query to be a JSON object where the from property defines the offset
+ * and the limit defines the page size.
+ * Expect the client to return a parsed JSON.
+ */
+'use strict';
+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
+ */
+function ReadableHits(queryExec, query, limit, emitSourceOnly) {
+  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._next = true;
+  if (limit > 0) {
+    this.limit = limit;
+  }
+
+// current iteration through the page
+  this._hits = [];
+  this._current = 0;
+}
+
+ReadableHits.prototype = Object.create(Readable.prototype, {constructor: {value: ReadableHits}});
+
+ReadableHits.prototype._read = function() {//size) {
+  this._current++;
+  if (this._current >= this._hits.length) {
+    if (!this._next) {
+      return this.push(null);
+    }
+    this._fetchNextPage();
+  } else {
+    this._shift();
+  }
+};
+
+ReadableHits.prototype._fetchNextPage = function() {//size) {
+  this.from += this.pageSize;
+  var self = this;
+  this.queryExec(this.query, function(e, resp) {
+    self._current = 0;
+    self._hits = resp.hits ? resp.hits.hits : resp.docs.docs;
+    if (self.from + self.pageSize > self._hits.length) {
+      self._next = false;
+    }
+    self._shift();
+  });
+};
+
+ReadableHits.prototype._shift = function() {//size) {
+  this.push(this._hits[this._current]);
+};

lib/writable-bulk.js

@@ -0,0 +1,103 @@
+/**
+ * Expose a writeable stream and execute it as a set of bulk requests.
+ */
+'use strict';
+
+var Writable = require('stream').Writable;
+
+module.exports = WritableBulk;
+
+/**
+ * @param bulkExec closure pass 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',
+ *                    _index: 'name of the index or nothing',
+ *                    _type:  'name of the type or nothing' }
+ *                nothing to emit an error on unknown document command
+ */
+function WritableBulk(bulkExec, defaults, bulkSize) {
+  Writable.call(this, {objectMode:true});
+  this.bulkExec = bulkExec;
+  this.bulkSize = bulkSize || 128;
+  this.defaults = defaults || {};
+  this.defaults.id = this.defaults.id || '_id';
+  this.defaults.op = this.defaults.op || 'index';
+
+  this.bulk = [];
+  this.bulkCount = 0;
+  this.expectingPayload = false;
+}
+
+WritableBulk.prototype = Object.create(Writable.prototype, {constructor: {value: WritableBulk}});
+
+/**
+ * @param chunk a piece of a bulk request as json.
+ */
+WritableBulk.prototype._write = function(chunk, enc, next) {
+  if (this.expectingPayload) {
+    this.bulkCount++;
+    this.expectingPayload = false;
+  } else if (chunk.hasOwnProperty(this.defaults.id)) {
+    var defaultCmd = {};
+    defaultCmd[this.defaults.op] = {
+      _index: this.defaults._index,
+      _type: this.defaults._type,
+      _id: chunk[this.defaults.id]
+    };
+    this.bulk.push(defaultCmd);
+    this.bulkCount++;
+  } else {
+    var willExpectPayload = ['index', 'create', 'update'];
+    for (var i = 0; i < willExpectPayload.length; i++) {
+      if (chunk.hasOwnProperty(willExpectPayload[i])) {
+        this.expectingPayload = willExpectPayload[i];
+        break;
+      }
+    }
+    if (!this.expectingPayload) {
+      if (!chunk.hasOwnProperty('delete')) {
+        this.emit('error', new Error('Unexpected chunk, not index create update delete'));
+        return next();
+      }
+      this.bulkCount++;
+    }
+  }
+  this.bulk.push(chunk);
+  if (this.bulkSize <= this.bulkCount) {
+    return this._flushBulk(next);
+  }
+  next();
+};
+
+WritableBulk.prototype._flushBulk = function(callback) {
+  if (!this.bulkCount) {
+    return setImmediate(callback);
+  }
+  var self = this;
+  this.bulkExec(this.bulk, function(e) {
+    if (e) {
+      // TODO: better than this?
+      // - Introspect the response for individual errors
+      // - Stream out the responses and correlate with the inputs?
+      self.emit('error', e);
+    }
+    self.bulk = [];
+    self.bulkCount = 0;
+    self.expectingPayload = false;
+    callback();
+  });
+};
+
+WritableBulk.prototype.end = function(data) {
+  var self = this;
+  if (!data) {
+    return this._flushBulk(function() {
+      self.emit('end');
+    });
+  }
+  this._write(data, 'json', function() {
+    self._flushBulk(function() {
+      self.emit('end');
+    });
+  });
+};

package.json

@@ -0,0 +1,25 @@
+{
+  "name": "elasticsearch-streams",
+  "description": "Stream in and out of Elasticsearch",
+  "version": "0.0.1",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/hmalphettes/elasticsearch-streamer.git"
+  },
+  "main": "index.js",
+  "scripts": {
+    "test": "jshint lib/*.js test/*.js && mocha"
+  },
+  "author": "Hugues Malphettes",
+  "license": "MIT",
+  "keywords": [
+    "elasticsearch",
+    "stream"
+  ],
+  "dependencies": {
+  },
+  "devDependencies": {
+    "chai": "*",
+    "elasticsearch": "*"
+  }
+}

test/test-search.js

@@ -0,0 +1,59 @@
+'use strict';
+var expect = require('chai').expect;
+var ReadableHits = require('../lib/readable-hits');
+var Writable = require('stream').Writable;
+var client = new require('elasticsearch').Client();
+
+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) {
+      if (e) { return done(e); }
+      var params = {
+        index: 'myindex',
+        from: 0,
+        size: 12,
+        body: {
+          query: { match_all: {} }
+        }
+      };
+      var queryExec = function(params, callback) {
+        client.search(params, callback);
+      };
+      rs = new ReadableHits(queryExec, params);
+      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);
+      done();
+    });
+    rs.pipe(ws);
+  });
+});