server/services/typeaheads/elastic-bulk-update-stream.js

   1 'use strict';
   2
   3 var Writable = require('stream').Writable;
   4 var util = require('util');
   5 var bulkTools = require('./elastic-bulk-tools');
   6 var debug = require('debug')('gitter:app:elastic-bulk-update-stream');
   7
   8 function ElasticBulkUpdateStream(elasticClient) {
   9   this._elasticClient = elasticClient;
  10   Writable.call(this, { objectMode: true });
  11 }
  12
  13 util.inherits(ElasticBulkUpdateStream, Writable);
  14
  15 // expects an array of updates where each element is a valid req for elasticClient.update
  16 ElasticBulkUpdateStream.prototype._write = function(updates, encoding, callback) {
  17   debug('uploading %d updates', updates.length);
  18   var req = bulkTools.createBulkUpdate(updates);
  19   this._elasticClient.bulk(req, function(err, res) {
  20     return callback(err || bulkTools.findErrors(req, res));
  21   });
  22 };
  23
  24 module.exports = ElasticBulkUpdateStream;