Merge pull request #11 from marshall007/custom-action

Support overriding the "action" for an operation

README.md

@@ -17,7 +17,8 @@ The records written to the stream has to have the following format:
   index: 'name-of-index',
   type: 'recordType',
   id: 'recordId',
-  parent: 'parentRecordType', //optional
+  parent: 'parentRecordType', // optional
+  action: 'update', // optional (default: 'index')
   body: {
     name: 'Foo Bar'
   }

index.js

@@ -18,23 +18,27 @@ function transformRecords(records) {
     return records.reduce(function(bulkOperations, record) {
         assert(record.index, 'index is required');
         assert(record.type, 'type is required');
-        assert(record.body, 'body is required');
 
-        var index = {
+        record.action = record.action || 'index';
+
+        var operation = {};
+
+        operation[record.action] = {
             _index: record.index,
             _type: record.type,
             _id: record.id
         };
 
         if (record.parent) {
-            index._parent = record.parent;
+            operation[record.action]._parent = record.parent;
         }
 
-        bulkOperations.push({
-            index: index
-        });
+        bulkOperations.push(operation);
 
-        bulkOperations.push(record.body);
+        if (record.action !== 'delete') {
+            assert(record.body, 'body is required');
+            bulkOperations.push(record.body);
+        }
 
         return bulkOperations;
     }, []);
@@ -83,11 +87,13 @@ ElasticsearchBulkIndexWritable.prototype.bulkWrite = function bulkWrite(records,
         }
 
         if (data.errors === true) {
-            var errors = data.items.map(function(item) {
+            var errors = _.chain(data.items)
+            .map(function(item) {
                 return _.map(item, 'error')[0];
-            });
-
-            errors = _.uniq(_.filter(errors, _.isString));
+            })
+            .filter(_.isString)
+            .uniq()
+            .value();
 
             if (this.logger) {
                 errors.forEach(this.logger.error.bind(this.logger));

test/elasticsearch-bulk-stream.js

@@ -11,9 +11,13 @@ chai.use(sinonChai);
 var expect = chai.expect;
 
 var recordFixture = require('./fixture/record.json');
-var recordParentFixture = require('./fixture/parentrecord.json');
+var recordDeleteFixture = require('./fixture/record-delete.json');
+var recordParentFixture = require('./fixture/record-parent.json');
+var recordUpdateFixture = require('./fixture/record-update.json');
 var successResponseFixture = require('./fixture/success-response.json');
+var successDeleteResponseFixture = require('./fixture/success-delete-response.json');
 var successParentResponseFixture = require('./fixture/success-parent-response.json');
+var successUpdateResponseFixture = require('./fixture/success-update-response.json');
 var errorResponseFixture = require('./fixture/error-response.json');
 
 describe('ElastisearchBulkIndexWritable', function() {
@@ -242,4 +246,62 @@ describe('ElastisearchBulkIndexWritable', function() {
             expect(this.client.bulk).to.have.been.calledWith(expectedArgument);
         });
     });
+
+    describe('custom action', function() {
+        beforeEach(function() {
+            this.client = {
+                bulk: this.sinon.stub()
+            };
+
+            this.stream = new ElasticsearchBulkIndexWritable(this.client, {
+                highWaterMark: 1,
+                flushTimeout: 10
+            });
+
+            this.clock = sinon.useFakeTimers();
+        });
+
+        it('should allow you to override the action', function() {
+            this.client.bulk.yields(null, successUpdateResponseFixture);
+            this.stream.write(recordUpdateFixture);
+
+            var expectedArgument = {
+                body: [
+                    {
+                        update: {
+                            _index: 'indexName',
+                            _type: 'recordType',
+                            _id: 'recordId'
+                        }
+                    },
+                    {
+                        foo: 'bar'
+                    }
+                ]
+            };
+
+            expect(this.client.bulk).to.have.callCount(1);
+            expect(this.client.bulk).to.have.been.calledWith(expectedArgument);
+        });
+
+        it('should not include body if action is delete', function() {
+            this.client.bulk.yields(null, successDeleteResponseFixture);
+            this.stream.write(recordDeleteFixture);
+
+            var expectedArgument = {
+                body: [
+                    {
+                        delete: {
+                            _index: 'indexName',
+                            _type: 'recordType',
+                            _id: 'recordId'
+                        }
+                    }
+                ]
+            };
+
+            expect(this.client.bulk).to.have.callCount(1);
+            expect(this.client.bulk).to.have.been.calledWith(expectedArgument);
+        });
+    });
 });

test/fixture/record-delete.json

@@ -0,0 +1,6 @@
+{
+    "index": "indexName",
+    "id": "recordId",
+    "type": "recordType",
+    "action": "delete"
+}

test/fixture/record-update.json

@@ -0,0 +1,9 @@
+{
+    "index": "indexName",
+    "id": "recordId",
+    "type": "recordType",
+    "action": "update",
+    "body": {
+        "foo": "bar"
+    }
+}

test/fixture/success-delete-response.json

@@ -0,0 +1,15 @@
+{
+    "took": 30,
+    "errors": false,
+    "items": [
+        {
+            "delete": {
+                "_index": "indexName",
+                "_id": "recordId",
+                "_type": "recordType",
+                "_version": 5,
+                "status": 200
+            }
+        }
+    ]
+}

test/fixture/success-update-response.json

@@ -0,0 +1,15 @@
+{
+    "took": 30,
+    "errors": false,
+    "items": [
+        {
+            "update": {
+                "_index": "indexName",
+                "_id": "recordId",
+                "_type": "recordType",
+                "_version": 5,
+                "status": 200
+            }
+        }
+    ]
+}