rethinkdbdash - npm Package Compare versions

Comparing version 2.1.2 to 2.1.3

lib/transform_stream.js

@@ -8,2 +8,3 @@ var Transform = require('stream').Transform;
   this._table = table;
+  this._r = table._r;
   this._options = options;
@@ -125,13 +126,32 @@ this._cache = [];
 
-  self._table.insert(cache, self._insertOptions).run(self._connection).then(function(result) {
+  var query = self._table.insert(cache, self._insertOptions);
+  if (self._options.format === 'primaryKey') {
+    query = query.do(function(result) {
+      return self._r.branch(
+        result('errors').eq(0),
+        self._table.config()('primary_key').do(function(primaryKey) {
+          return result('changes')('new_val')(primaryKey)
+        }),
+        result(self._r.error(result('errors').coerceTo('STRING').add(' errors returned. First error:\n').add(result('first_error'))))
+      )
+    })
+  }
+
+  query.run(self._connection).then(function(result) {
     self._inserting = false;
-
-    if (result.errors > 0) {
-      self._inserting = false;
-      self.emit('error', new Error('Failed to insert some documents:'+JSON.stringify(result, null, 2)));
+    if (self._options.format === 'primaryKey') {
+      for(var i=0; i<result.length; i++) {
+        self.push(result[i]);
+      }
     }
     else {
-      if (self._insertOptions.returnChanges === true) {
-        for(var i=0; i<result.changes.length; i++) {
-          self.push(result.changes[i].new_val);
+      if (result.errors > 0) {
+        self._inserting = false;
+        self.emit('error', new Error('Failed to insert some documents:'+JSON.stringify(result, null, 2)));
+      }
+      else {
+        if (self._insertOptions.returnChanges === true) {
+          for(var i=0; i<result.changes.length; i++) {
+            self.push(result.changes[i].new_val);
+          }
         }
@@ -138,0 +158,0 @@ }

package.json

 {
   "name": "rethinkdbdash",
-  "version": "2.1.2",
+  "version": "2.1.3",
   "description": "A Node.js driver for RethinkDB with promises and a connection pool",
@@ -5,0 +5,0 @@ "main": "lib/index.js",

README.md

@@ -242,2 +242,4 @@ rethinkdbdash
 
+_Note_: The option `{optionalRun: false}` will disable the optional run for all instances of the driver.
+
 #### Connection pool
@@ -371,3 +373,4 @@
 
-A transform stream will return the saved documents.
+By default, a transform stream will return the saved documents. You can return the primary
+key of the new document by passing the option `format: 'primaryKey'`.
 
@@ -374,0 +377,0 @@ This makes a convenient way to dump a file your database.

test/transform-stream.js

@@ -6,2 +6,3 @@ var config = require('./config.js');
 var Readable = require('stream').Readable;
+var Stream = require('stream')
 var _util = require('util');
@@ -179,1 +180,102 @@ var devnull = require('dev-null');
 })
+
+It('test transform output - object', function* (done) {
+  var stream = new Readable({objectMode: true});
+  var i = 0;
+  var values = [uuid(), uuid()];
+  stream._read = function() {
+    var self = this;
+    i++;
+    if (i <= 3) {
+      self.push({field: values[0]});
+    }
+    else if (i === 4) {
+      setTimeout(function() {
+        self.push({field: values[1]});
+      }, 300)
+    }
+    else if (i <= 10) {
+      self.push({field: values[1]});
+    }
+    else {
+      self.push(null);
+    }
+  }
+
+  var table = r.db(dbName).table(dumpTable).toStream({
+    transform: true
+  });
+
+  var result = [];
+  var endStream = new Stream.Transform();
+  endStream._writableState.objectMode = true;
+  endStream._readableState.objectMode = true;
+  endStream._transform = function (data, encoding, done) {
+    result.push(data);
+    this.push(data);
+    done();
+  }
+
+  stream.pipe(table)
+    .on('error', done)
+    .pipe(endStream)
+    .on('error', done)
+    .on('finish', function() {
+      assert(result.length, 10);
+      for(var i=0; i<result.length; i++) {
+        assert(Object.prototype.toString.call(result[i]), '[object Object]');
+      }
+      done();
+    });
+})
+
+It('test transform output - string', function* (done) {
+  var stream = new Readable({objectMode: true});
+  var i = 0;
+  var values = [uuid(), uuid()];
+  stream._read = function() {
+    var self = this;
+    i++;
+    if (i <= 3) {
+      self.push({field: values[0]});
+    }
+    else if (i === 4) {
+      setTimeout(function() {
+        self.push({field: values[1]});
+      }, 300)
+    }
+    else if (i <= 10) {
+      self.push({field: values[1]});
+    }
+    else {
+      self.push(null);
+    }
+  }
+
+  var table = r.db(dbName).table(dumpTable).toStream({
+    transform: true,
+    format: 'primaryKey'
+  });
+
+  var result = [];
+  var endStream = new Stream.Transform();
+  endStream._writableState.objectMode = true;
+  endStream._readableState.objectMode = true;
+  endStream._transform = function (data, encoding, done) {
+    result.push(data);
+    this.push(data);
+    done();
+  }
+
+  stream.pipe(table)
+    .on('error', done)
+    .pipe(endStream)
+    .on('error', done)
+    .on('finish', function() {
+      assert(result.length, 10);
+      for(var i=0; i<result.length; i++) {
+        assert(typeof result[i], 'string');
+      }
+      done();
+    });
+})

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

816994

increased by0.4%

Lines of code

24581

increased by0.47%

Number of lines in readme file

525

increased by0.57%

No dependency changes