elasticsearch-stream-index - npm Package Compare versions

Comparing version 1.0.1 to 1.0.2

index.js

@@ -10,3 +10,3 @@ // node modules
 try{
-  debug = require('debug')('StreamingElasticsearch:BulkStream');
+  debug = require('debug')('elasticsearch-stream-index:BulkStream');
 } catch (err){}
@@ -18,5 +18,5 @@
   }
-  Writable.call(this, { objectMode: true });
+  this.highWaterMark = (options && options.highWaterMark) || 64;
+  Writable.call(this, { objectMode: true, highWaterMark: this.highWaterMark });
   this.es_client = es_client;
-  this.highWaterMark = (options && options.highWaterMark) || 64;
   this.bulkCount = 0;
@@ -54,3 +54,3 @@ this.arr = [];
   });
-  debug('_bulk bufArray', bufArray.length);
+  // debug('_bulk bufArray', bufArray.length);
   // bulk insert
@@ -63,5 +63,13 @@ this.es_client.bulk({ body: bufArray }, function(err, resp){
     debug('_bulk resp', resp);
+
+     // add fail handle
+    if (resp.errors === true) {
+      debug('_bulk error', resp.items);
+      self.emit('fail', resp.items);
+    }
+
     self.bulkCount = 0;
     self.nextChunkIsParams = false;
     self.arr = [];
+
     callback();
@@ -68,0 +76,0 @@ });

package.json

 {
   "name": "elasticsearch-stream-index",
-  "version": "1.0.1",
+  "version": "1.0.2",
   "description": "a writable stream wrapped elasticsearch index operation to bulk",
   "main": "index.js",
   "scripts": {
-    "test": "node test/test.js"
+    "test": "DEBUG=* node test/test.js"
   },
@@ -20,4 +20,4 @@ "repository": {
   "dependencies": {
-    "elasticsearch": "^3.1.3",
-    "readable-stream": "^1.0.33"
+    "elasticsearch": "^11.0.0",
+    "readable-stream": "^2.0.6"
   },
@@ -24,0 +24,0 @@ "devDependencies": {

README.md

@@ -1,2 +0,2 @@
-# ElasticsearchStreamIndex ![io.js supported](https://img.shields.io/badge/io.js-supported-green.svg?style=flat) [![Build Status](https://travis-ci.org/topdmc/ElasticsearchStreamIndex.svg?branch=master)](https://travis-ci.org/topdmc/ElasticsearchStreamIndex)
+# ElasticsearchStreamIndex ![io.js supported](https://img.shields.io/badge/io.js-supported-green.svg?style=flat) [![Build Status](https://travis-ci.org/topdmc/ElasticsearchStreamIndex.svg?branch=master)](https://travis-ci.org/topdmc/ElasticsearchStreamIndex) [![npm version](https://badge.fury.io/js/elasticsearch-stream-index.svg)](http://badge.fury.io/js/elasticsearch-stream-index)
 
@@ -12,4 +12,3 @@ A writable stream wrapped elasticsearch index operation to bulk
 ```javascript
-var ElasticsearchStreamIndex = require('../index');
-
+var ElasticsearchStreamIndex = require('elasticsearch-stream-index');
 var elasticsearch = require('elasticsearch');
@@ -22,22 +21,53 @@
 
-var es = new elasticsearch.Client(opt);
+var idxName = 'cc_idx', typeName = 'cc_type';
 
-var create_stream = new ElasticsearchStreamIndex(es, { highWaterMark: 2 });
-create_stream.on('finish', function(){
-  console.log('---finished---');
+// 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 idxName = 'cc_idx', typeName = 'cc_type';
+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); });
 
-for(var i = 0; i < 10; i++){
-  create_stream.write({
-    index: idxName,
-    type: typeName,
-    id: i,
-    body : {
-      name: 'name _ ' + i
-    }
-  });
-}
-create_stream.end();
+rs.pipe(esi);
 ```

test/test.js

 var assert = require('assert');
+var stream = require('stream');
+var util = require('util');
 
@@ -6,29 +8,79 @@ var ElasticsearchStreamIndex = require('../index');
 var elasticsearch = require('elasticsearch');
+var idxName = 'test', typeName = 'cc_type';
 
-var opt = {
-  host: 'localhost:9200',
-  log: 'error'
+/* Generate a readable-stream demostrate data */
+const Readable = require('stream').Readable;
+util.inherits(Counter, Readable);
+
+function Counter(opt) {
+  Readable.call(this, opt);
+  this._max = 50;
+  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 es = new elasticsearch.Client(opt);
+/* Generate a transform stream */
+function StringifyStream(options) {
+  if (!(this instanceof StringifyStream))
+    return new StringifyStream(options);
 
-var create_stream = new ElasticsearchStreamIndex(es, { highWaterMark: 2 });
-create_stream.on('finish', function(){
-  assert.ok(true);
+  options = options || {};
+  options.objectMode = true;
+
+  stream.Transform.call(this,options);
+}
+
+util.inherits(StringifyStream,stream.Transform);
+
+StringifyStream.prototype._transform = function(d,e,callback) {
+  this.push(JSON.stringify(d));
+  this.push("\n");
+  callback();
+};
+
+/* try */
+var rs = new Counter({ objectMode: true });
+rs.on('readable', function(){
+  console.log(">>> rs.readable");
 });
+// rs.pipe(StringifyStream()).pipe(process.stdout);
 
-create_stream.on('error', function(){ assert.ok(false); });
+/* Create the writer */
+var es = new elasticsearch.Client({
+  host: 'localhost:9200',
+  log: 'error'
+});
 
-var idxName = 'cc_idx', typeName = 'cc_type';
+var esi = new ElasticsearchStreamIndex(es, { highWaterMark: 7 });
+esi.on('finish', function(){
+  assert.ok(true);
+});
+esi.on('pipe', function(src){
+  console.log('>> something is piping into me');
+  assert.equal(src, rs);
+});
+esi.on('unpipe', function(src){
+  console.log('>> something is unpiping into me');
+  assert.equal(src, rs);
+});
+esi.on('error', function(){ assert.ok(false); });
 
-for(var i = 0; i < 10; i++){
-  create_stream.write({
-    index: idxName,
-    type: typeName,
-    id: i,
-    body : {
-      name: 'name _ ' + i
-    }
-  });
-}
-create_stream.end();
+/* Testing */
+rs.pipe(esi);
+// esi.end();

No alert changes

Improved metrics

Total package byte prevSize

8145

40.12%

Lines of code

155

49.04%

Number of lines in readme file

72

71.43%

Dependency changes

• + Added

  ansi-regex@2.1.1
  (transitive)

• + Added

  ansi-styles@2.2.1
  (transitive)

• + Added

  asap@2.0.6
  (transitive)

• + Added

  chalk@1.1.3
  (transitive)

• + Added

  elasticsearch@11.0.1
  (transitive)

• + Added

  forever-agent@0.6.1
  (transitive)

• + Added

  has-ansi@2.0.0
  (transitive)

• + Added

  isarray@1.0.0
  (transitive)

• + Added

  lodash-compat@3.10.2
  (transitive)

• + Added

  process-nextick-args@2.0.1
  (transitive)

• + Added

  promise@7.3.1
  (transitive)

• + Added

  readable-stream@2.3.8
  (transitive)

• + Added

  safe-buffer@5.1.2
  (transitive)

• + Added

  string_decoder@1.1.1
  (transitive)

• + Added

  strip-ansi@3.0.1
  (transitive)

• + Added

  supports-color@2.0.0
  (transitive)

• + Added

  util-deprecate@1.0.2
  (transitive)

• - Removed

  ansi-regex@0.2.1
  (transitive)

• - Removed

  ansi-styles@1.1.0
  (transitive)

• - Removed

  bluebird@2.8.2
  (transitive)

• - Removed

  chalk@0.5.1
  (transitive)

• - Removed

  elasticsearch@3.1.4
  (transitive)

• - Removed

  forever-agent@0.5.2
  (transitive)

• - Removed

  has-ansi@0.1.0
  (transitive)

• - Removed

  isarray@0.0.1
  (transitive)

• - Removed

  lodash-node@2.4.1
  (transitive)

• - Removed

  readable-stream@1.1.14
  (transitive)

• - Removed

  string_decoder@0.10.31
  (transitive)

• - Removed

  strip-ansi@0.3.0
  (transitive)

• - Removed

  supports-color@0.2.0
  (transitive)

• Updated

  elasticsearch@^11.0.0

• Updated

  readable-stream@^2.0.6