Support Upserts to ElasticSearch

lib/elasticsearchclient/calls/core.js

@@ -237,4 +237,33 @@ ElasticSearchClient.prototype.update = function(indexName, typeName, documentId,
         path += "?" + qs;
     }
     return this.createCall({data: JSON.stringify(document), path: path, method: 'POST'}, this.clientOptions);
+}
+
+ElasticSearchClient.prototype.upsert = function(indexName, typeName, documentId, script, params, document, options) {
+    document = document || {};
+    // if no record exists for this id, it will be inserted based on the document. 
+    var command = {
+		upsert: document
+    };
+    // if the document for the given ID already exists, the script will be run against it,
+    // so put any update calls in here:
+    if (script) {
+    	command.script = script;
+    }
+    // params used by the script
+    if (params) {
+    	command.params = params;
+    }
+    if(!documentId)
+        throw "id should be provided for upsert: " + JSON.stringify(command);
+
+    var path = '/' + indexName + '/' + typeName + '/'+documentId+'/_update';
+    var qs = '';
+    if (options) {
+        qs = querystring.stringify(options)
+    }
+    if (qs.length > 0) {
+        path += "?" + qs;
+    }
+    return this.createCall({data: JSON.stringify(command), path: path, method: 'POST'}, this.clientOptions);
 }

lib/elasticsearchclient/calls/elasticSearchCall.js

@@ -57,7 +57,9 @@ ElasticSearchCall.prototype.exec = function() {
     if (this.auth) {
         request.setHeader("Authorization", "Basic " + new Buffer(this.auth.username + ":" + this.auth.password).toString('base64'))
     }
-
+    // BA - try keep alive
+    request.setHeader('Connection', 'keep-alive');
+
     if (this.params.data) {
         if (typeof this.params.data != 'string') {
             this.params.data = JSON.stringify(this.params.data);

test/core.test.js

@@ -237,4 +237,62 @@ describe("ElasticSearchClient Core api", function(){
                 .exec();
          });
     });
+
+    describe("#upsert", function(){
+    	var id = "upsertable";
+    	var initialValue = 'first';
+    	var secondValue = 'second';
+        it("should insert non existing document", function(done){
+        	var script = "ctx._source.occupation = param1;";
+        	var params = {param1: "doesn't matter as this shouldn't apply for a new document"};
+        	var doc = {occupation: initialValue};
+            elasticSearchClient.upsert(indexName, objName, id , script, params, doc)
+            .on('data', function(data) {
+                data = JSON.parse(data);
+                data.should.be.ok;
+                data._id.should.equal(id);
+                done();
+            })
+            .exec();
+        });
+        it("should fetch the newly upserted row by id", function(done){
+            elasticSearchClient.get(indexName, objName, id)
+            .on('data', function(data) {
+                data = JSON.parse(data);
+                data.exists.should.exist;
+                data._id.should.equal(id);
+                // ensure that the insert part of the upsert has worked as this should be a new record.
+                data._source.occupation.should.equal(initialValue);
+                done();
+            })
+            .exec();
+        });        
+        it("should update existing document", function(done){
+        	var id = "upsertable";
+        	var script = 'ctx._source.occupation = param1;';
+        	var params = {param1: secondValue};
+        	var doc = {occupation: "doesn't matter as this shouldn't be applied to existing document"};
+        	elasticSearchClient.upsert(indexName, objName,id , script, params, doc)
+        	.on('data', function(data) {
+        		data = JSON.parse(data);
+        		data.should.be.ok;
+        		data._id.should.equal(id);
+        		done();
+        	})
+        	.exec()
+        });
+        it("should fetch the updated row by id", function(done){
+            elasticSearchClient.get(indexName, objName, id)
+            .on('data', function(data) {
+                data = JSON.parse(data);
+                data.exists.should.exist;
+                data._id.should.equal(id);
+                // make sure that the script has executed to update the existing record
+                data._source.occupation.should.equal(secondValue);
+                done();
+            })
+            .exec();
+        });  
+    });    
+
 });