ElasticsearchStreamIndex

A writable stream wrapped elasticsearch index operation to bulk

Install

npm install elasticsearch-stream-index

Usage

var ElasticsearchStreamIndex = require('elasticsearch-stream-index');
var elasticsearch = require('elasticsearch');

var opt = {
  host: 'localhost:9200',
  log: 'error'
};

var idxName = 'cc_idx', typeName = 'cc_type';

// readable-stream
const Readable = require('stream').Readable;
util.inherits(Counter, Readable);

function Counter(opt) {
  Readable.call(this, opt);
  this._max = 1000;
  this._index = 0;
}

Counter.prototype._read = function() {
  var i = this._index++;
  console.log('---| _read i=' + i);
  if (i > this._max)
    this.push(null);
  else {
    this.push({
      index: idxName,
      type: typeName,
      id: i,
      body : {
        name: 'name _ ' + i
      }
    });
  }
};

var rs = new Counter({ objectMode: true });

var es = new elasticsearch.Client({
  host: 'localhost:9200',
  log: 'error'
});

var esi = new ElasticsearchStreamIndex(es, { highWaterMark: 128 });
esi.on('finish', function(){
  assert.ok(true);
});
esi.on('pipe', function(src){
  console.error('>> something is piping into me');
  assert.equal(src, rs);
});
esi.on('unpipe', function(src){
  console.error('>> something is unpiping into me');
  assert.equal(src, rs);
});
esi.on('error', function(){ assert.ok(false); });

rs.pipe(esi);