Add write and mWrite

lib/api/controllers/bulkController.js

@@ -23,16 +23,27 @@
 
 const
   BaseController = require('./controller'),
-  {errors: {PartialError}} = require('kuzzle-common-objects'),
-  {assertHasBody, assertBodyHasAttribute} = require('../../util/requestAssertions');
+  { PartialError } = require('kuzzle-common-objects').errors,
+  {
+    assertHasBody,
+    assertBodyHasAttribute,
+    assertHasIndexAndCollection,
+    assertBodyAttributeType,
+    assertIdStartsNotUnderscore
+  } = require('../../util/requestAssertions');
 
 /**
  * @class BulkController
  * @param {Kuzzle} kuzzle
  */
 class BulkController extends BaseController {
   constructor (kuzzle) {
-    super(kuzzle, ['import']);
+    super(kuzzle, [
+      'import',
+      'write',
+      'mWrite'
+    ]);
+
     this.engine = kuzzle.services.list.storageEngine;
   }
 
@@ -55,6 +66,63 @@ class BulkController extends BaseController {
         return response;
       });
   }
+
+  /**
+   * Write a document without adding metadata or performing data validation.
+   * @param {Request} request
+   * @returns {Promise<Object>}
+   */
+  write(request) {
+    assertHasBody(request);
+    assertHasIndexAndCollection(request);
+    assertIdStartsNotUnderscore(request);
+
+    const notify = this.ensureBooleanFlag(request, 'input.args.notify');
+
+    return this.engine.writeDocument(request)
+      .then(response => {
+        this.kuzzle.indexCache.add(request.input.resource.index, request.input.resource.collection);
+
+
+        if (notify && response.created) {
+          this.kuzzle.notifier.notifyDocumentCreate(request, response);
+        }
+        else if (notify) {
+          this.kuzzle.notifier.notifyDocumentReplace(request);
+        }
+
+        return response;
+      });
+  }
+
+  /**
+   * Write several documents without adding metadata or performing data validation.
+   *
+   * @param {Request} request
+   * @returns {Promise<Object>}
+   */
+  mWrite(request) {
+    assertHasBody(request);
+    assertBodyHasAttribute(request, 'documents');
+    assertBodyAttributeType(request, 'documents', 'array');
+    assertHasIndexAndCollection(request);
+
+    const notify = this.ensureBooleanFlag(request, 'input.args.notify');
+
+    return this.engine.mwriteDocuments(request)
+      .then(response => {
+        if (response.error.length > 0) {
+          request.setError(new PartialError('Some document creations failed', response.error));
+        }
+
+        if (notify) {
+          this.kuzzle.notifier.notifyDocumentMChanges(request, response.result, true);
+        }
+
+        return {hits: response.result, total: response.result.length};
+      });
+  }
+
 }
 
 module.exports = BulkController;

lib/api/controllers/controller.js

@@ -82,8 +82,11 @@ class BaseController {
       const booleanValue = flagValue !== undefined ? true : false;
 
       _.set(request, flagPath, booleanValue);
+
+      return booleanValue;
     }
 
+    return false;
   }
 }
 

lib/services/elasticsearch.js

@@ -374,6 +374,34 @@ class ElasticSearch extends Service {
       });
   }
 
+  /**
+   * Create a new document to ElasticSearch, or replace it if it already exist.
+   * This method does not adds additional metadata.
+   *
+   * @param {Request} request
+   * @returns {Promise} resolve an object that contains _id
+   */
+  writeDocument(request) {
+    const esRequest = initESRequest(request, ['refresh']);
+
+    esRequest.id = request.input.resource._id;
+    esRequest.body = request.input.body;
+
+    assertNoRouting(esRequest);
+    assertWellFormedRefresh(esRequest);
+
+    return this.kuzzle.indexCache.exists(esRequest.index, esRequest.type)
+      .then(exists => {
+        if (! exists) {
+          throw new PreconditionError(`Index '${esRequest.index}' and/or collection '${esRequest.type}' does not exist`);
+        }
+
+        return this.client.index(esRequest)
+          .then(result => this.refreshIndexIfNeeded(esRequest, _.extend(result, {_source: request.input.body})))
+          .catch(error => Bluebird.reject(this.esWrapper.formatESError(error)));
+      });
+  }
+
   /**
    * Sends the partial document to elasticsearch
    * with the id to update
@@ -1109,6 +1137,35 @@ class ElasticSearch extends Service {
     return this._mexecute(esRequest, extracted.documents, extracted.rejected);
   }
 
+  /**
+   * Create or replace multiple documents at once.
+   * Does not adds additional metadata
+   *
+   * @param  {Request} request - Kuzzle API request
+   * @return {Promise}
+   */
+  mwriteDocuments(request) {
+    const
+      esRequest = initESRequest(request, ['consistency', 'refresh', 'timeout', 'fields']),
+      extracted = extractMDocuments(request, {});
+
+    esRequest.body = [];
+
+    let i; // NOSONAR
+    for (i = 0; i < extracted.documents.length; i++) {
+      esRequest.body.push({
+        index: {
+          _index: request.input.resource.index,
+          _type: request.input.resource.collection,
+          _id: extracted.documents[i]._id
+        }
+      });
+      esRequest.body.push(extracted.documents[i]._source);
+    }
+
+    return this._mexecute(esRequest, extracted.documents, extracted.rejected);
+  }
+
   /**
    * Update multiple documents with one request
    * Replacements are rejected if targeted documents do not exist,
@@ -1284,7 +1341,7 @@ class ElasticSearch extends Service {
   }
 
   /**
-   * Execute an ES request prepared by mcreate, mupdate, mreplace or mdelete
+   * Execute an ES request prepared by mcreate, mupdate, mreplace, mdelete or mwriteDocuments
    * Returns a standardized ES response object, containing the list of
    * successfully performed operations, and the rejected ones
    *