Merge branch 'add-parent-property'

voldern · Jul 22, 2016 · 5b66fc8 · 5b66fc8
2 parents 2a2bebd + b8219e7
commit 5b66fc8
Show file tree

Hide file tree

Showing 5 changed files with 78 additions and 5 deletions.
diff --git a/README.md b/README.md
@@ -17,6 +17,7 @@ The records written to the stream has to have the following format:
   index: 'name-of-index',
   type: 'recordType',
   id: 'recordId',
+  parent: 'parentRecordType', //optional
   body: {
     name: 'Foo Bar'
   }

diff --git a/index.js b/index.js
@@ -20,12 +20,18 @@ function transformRecords(records) {
         assert(record.type, 'type is required');
         assert(record.body, 'body is required');
 
+        var index = {
+            _index: record.index,
+            _type: record.type,
+            _id: record.id
+        };
+
+        if (record.parent) {
+            index._parent = record.parent;
+        }
+
         bulkOperations.push({
-            index: {
-                _index: record.index,
-                _type: record.type,
-                _id: record.id
-            }
+            index: index
         });
 
         bulkOperations.push(record.body);

diff --git a/test/elasticsearch-bulk-stream.js b/test/elasticsearch-bulk-stream.js
@@ -11,7 +11,9 @@ chai.use(sinonChai);
 var expect = chai.expect;
 
 var recordFixture = require('./fixture/record.json');
+var recordParentFixture = require('./fixture/parentrecord.json');
 var successResponseFixture = require('./fixture/success-response.json');
+var successParentResponseFixture = require('./fixture/success-parent-response.json');
 var errorResponseFixture = require('./fixture/error-response.json');
 
 describe('ElastisearchBulkIndexWritable', function() {
@@ -201,4 +203,43 @@ describe('ElastisearchBulkIndexWritable', function() {
             }.bind(this));
         });
     });
+
+    describe('parent type', function() {
+        beforeEach(function() {
+            this.client = {
+                bulk: this.sinon.stub()
+            };
+
+            this.stream = new ElasticsearchBulkIndexWritable(this.client, {
+                highWaterMark: 1,
+                flushTimeout: 10
+            });
+
+            this.client.bulk.yields(null, successParentResponseFixture);
+            this.clock = sinon.useFakeTimers();
+        });
+
+        it('should include parent type in record if present', function() {
+            this.stream.write(recordParentFixture);
+
+            var expectedArgument = {
+                body: [
+                    {
+                        index: {
+                            _index: 'indexName',
+                            _type: 'recordType',
+                            _id: 'recordId',
+                            _parent: 'parentRecordType'
+                        }
+                    },
+                    {
+                        foo: 'bar'
+                    }
+                ]
+            };
+
+            expect(this.client.bulk).to.have.callCount(1);
+            expect(this.client.bulk).to.have.been.calledWith(expectedArgument);
+        });
+    });
 });
diff --git a/test/fixture/parentrecord.json b/test/fixture/parentrecord.json
@@ -0,0 +1,9 @@
+{
+    "index": "indexName",
+    "id": "recordId",
+    "type": "recordType",
+    "parent": "parentRecordType",
+    "body": {
+        "foo": "bar"
+    }
+}
diff --git a/test/fixture/success-parent-response.json b/test/fixture/success-parent-response.json
@@ -0,0 +1,16 @@
+{
+    "took": 30,
+    "errors": false,
+    "items": [
+        {
+            "index": {
+                "_index": "indexName",
+                "_id": "recordId",
+                "_type": "recordType",
+                "_parent": "parentRecordType",
+                "_version": 5,
+                "status": 200
+            }
+        }
+    ]
+}