Emit errors on bulk and finish event instead of end

Author: hmalphettes

lib/writable-bulk.js

@@ -63,6 +63,7 @@ WritableBulk.prototype._write = function(chunk, enc, next) {
     }
     if (!this.expectingPayload) {
       if (!chunk.hasOwnProperty('delete')) {
+        console.log('humf', chunk);
         this.emit('error', new Error('Unexpected chunk, not an ' +
             'index/create/update/delete command and ' +
             'not a document to index either'));
@@ -83,13 +84,19 @@ WritableBulk.prototype._flushBulk = function(callback) {
     return setImmediate(callback);
   }
   var self = this;
-  this.bulkExec(this.bulk, function(e) {
+  this.bulkExec(this.bulk, function(e, resp) {
     if (e) {
-      // TODO: better than this?
-      // - Introspect the response for individual errors
-      // - Stream out the responses and correlate with the inputs?
       self.emit('error', e);
     }
+    if (resp.errors && resp.items) {
+      for (var i = 0; i < resp.items.length; i++) {
+        var bulkItemResp = resp.items[i];
+        var key = Object.keys(bulkItemResp)[0];
+        if (bulkItemResp[key].error) {
+          self.emit('error', new Error(bulkItemResp[key].error));
+        }
+      }
+    }
     self.bulk = [];
     self.bulkCount = 0;
     self.expectingPayload = false;
@@ -101,12 +108,12 @@ WritableBulk.prototype.end = function(data) {
   var self = this;
   if (!data) {
     return this._flushBulk(function() {
-      self.emit('end');
+      self.emit('finish');
     });
   }
   this._write(data, 'json', function() {
     self._flushBulk(function() {
-      self.emit('end');
+      self.emit('finish');
     });
   });
 };

test/test-write.js

@@ -20,8 +20,7 @@ describe('When writing', function() {
     ws = new WritableBulk(bulkExec);
     ws.on('error', function(e) {
       err = e;
-    });
-    ws.on('end', function() {
+    }).on('finish', function() {
       done(err);
     });
     // drop the index then