WriteableBulk accepts a stream of bulk cmds only

before it would be configurable to generate a default command
when it was passed a document.
It was mixing concern.

An example transformer stream that generates bulk commands to index
documents was added as `require('elasticsearch-streams').TransformToBulk

Author: hmalphettes

README.md

@@ -25,9 +25,9 @@ var bulkExec = function(bulkCmds, callback) {
   }, callback);
 };
 var ws = new WritableBulk(bulkExec);
-
+var toBulk = new TransformToBulk(function getIndexTypeId(doc) { return { _id: doc.id }; });
 // stream 42 random records into ES
-require('random-document-stream')(42).pipe(ws).on('finish', done);
+require('random-document-stream')(42).pipe(toBulk).pipe(ws).on('finish', done);
 ```
 
 ## Stream search results from Elasticsearch

index.js

@@ -1,4 +1,5 @@
 module.exports = {
+  TransformToBulk: require('./lib/transform-to-bulk.js'),
   WritableBulk   : require('./lib/writable-bulk'),
   ReadableSearch : require('./lib/readable-search'),
   PipableDocs    : require('./lib/transform-mget')

lib/transform-to-bulk.js

@@ -0,0 +1,28 @@
+/**
+ * Transform a stream of documents into a stream of bulk commands to index them.
+ * This is more an example than anything else.
+ */
+'use strict';
+var Transform = require('stream').Transform;
+
+module.exports = TransformToBulk;
+
+/**
+ * @param getIndexTypeId function that is passed a document and returns:
+ *            { _index: the_index?, _type: the_type?, _id: the_id? }
+ */
+function TransformToBulk(getIndexTypeId) {
+  Transform.call(this, {objectMode:true});
+  this.getIndexTypeId = getIndexTypeId;
+}
+
+TransformToBulk.prototype = Object.create(Transform.prototype, {constructor: {value: TransformToBulk}});
+
+TransformToBulk.prototype._transform = function(chunk, encoding, callback) {
+  var params = this.getIndexTypeId(chunk);
+  if (params) {
+    this.push({ index: params });
+    this.push(chunk);
+  }
+  callback();
+};

lib/writable-bulk.js

@@ -9,26 +9,13 @@ module.exports = WritableBulk;
 
 /**
  * @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',
- *                    _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.
+ * @param highWaterMark number of bulk commands executed at once. 128 by default.
  */
-function WritableBulk(bulkExec, defaults, bulkSize) {
+function WritableBulk(bulkExec, highWaterMark) {
   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';
-  this.defaults.op = this.defaults.op || 'index';
+  this.highWaterMark = highWaterMark || 128;
 
   this.bulk = [];
   this.bulkCount = 0;
@@ -44,15 +31,6 @@ 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++) {
@@ -73,7 +51,7 @@ WritableBulk.prototype._write = function(chunk, enc, next) {
     }
   }
   this.bulk.push(chunk);
-  if (this.bulkSize <= this.bulkCount) {
+  if (this.highWaterMark <= this.bulkCount) {
     return this._flushBulk(next);
   }
   next();

test/test-write.js

@@ -1,15 +1,16 @@
 'use strict';
 var expect = require('chai').expect;
 var WritableBulk = require('..').WritableBulk;
+var TransformToBulk = require('..').TransformToBulk;
 var random = require('random-document-stream');
-var client = new require('elasticsearch').Client();
+var client = new require('elasticsearch').Client({log: 'warning'});
 
 describe('When writing', function() {
   var ws;
   before(function(done) {
     var bulkExec = function(bulkCmds, callback) {
       client.bulk({
-        index : 'myindex',
+        index : 'myindex2',
         type  : 'mytype',
         body  : bulkCmds
       }, callback);
@@ -23,16 +24,20 @@ describe('When writing', function() {
     }).on('finish', function() {
       done(err);
     });
+
+    var transformToBulk = new TransformToBulk(function(doc) {
+      return { _id: doc._id };
+    });
     // drop the index then
     client.indices.delete({index: 'myindex2'}, function() {
       // stream 42 random docs into ES
-      random(42).pipe(ws);
+      random(42).pipe(transformToBulk).pipe(ws);
     });
   });
   it('Must have indexed 42 docs', function(done) {
     client.indices.refresh({ index: 'myindex2' }, function() {
       client.count({
-        index: 'myindex',
+        index: 'myindex2',
         type: 'mytype'
       }, function(e, res) {
         if (e) { return done(e); }