Support to parse the hit inside the ReadableStream

Sometimes it is too much a bother to setup a Transformer
stream just to execute a simple transformation of
the hits returned by Elasticsearch.

Example:
```
var num = 1;
new ReadableStream(bulkExec, function parseHit(hit) {
  return { id: hit._id, version: hit._version, hitNum: num++ };
});
```

Author: hmalphettes

README.md

@@ -89,7 +89,7 @@ var scrollExec = function scrollExec(from, callback) {
   client.search({
     index: 'myindex',
     scroll: '20s',
-    size: '3',
+    size: 42,
     body: {
       query: { match_all: {} }
     }

lib/readable-search.js

@@ -13,7 +13,7 @@ module.exports = ReadableHits;
 /**
  * @param queryExec an executable query functions that takes 2 arguments: the offset and a callback.
  */
-function ReadableHits(queryExec) {
+function ReadableHits(queryExec, parseHit) {
   if (!(this instanceof ReadableHits)) {
     return new ReadableHits(queryExec);
   }
@@ -26,6 +26,7 @@ function ReadableHits(queryExec) {
 // current iteration through the page
   this._hits = [];
   this._current = 0;
+  this.parseHit = parseHit || identity;
 }
 
 ReadableHits.prototype = Object.create(Readable.prototype, {constructor: {value: ReadableHits}});
@@ -68,5 +69,9 @@ ReadableHits.prototype._fetchNextPage = function() {//size) {
 };
 
 ReadableHits.prototype._shift = function() {
-  this.push(this._hits[this._current]);
+  this.push(this.parseHit(this._hits[this._current]));
 };
+
+function identity(hit) {
+  return hit;
+}

lib/writable-bulk.js

@@ -44,7 +44,6 @@ WritableBulk.prototype._write = function(chunk, enc, next) {
     }
     if (!this.expectingPayload) {
       if (!chunk.hasOwnProperty('delete')) {
-        console.log('humf', chunk);
         this.emit('error', new Error('Unexpected chunk, not an ' +
             'index/create/update/delete command and ' +
             'not a document to index either'));

package.json

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