jetstream - npm Package Compare versions

Comparing version 0.10.0 to 0.11.0

lib/query/abstract_query_operation.js

@@ -285,3 +285,3 @@ // Copyright (c) 2015 Uber Technologies, Inc.
 
-AbstractQueryOperation.prototype._applySyncFragmentsAndCreateQueryResult = function(scope, syncFragments, options, modelObject, callback) {
+AbstractQueryOperation.prototype._applySyncFragmentsWithLockAndCreateQueryResult = function(scope, syncFragments, options, modelObject, callback) {
     syncFragments = syncFragments.filter(function(syncFragment) {
@@ -294,3 +294,3 @@ if (syncFragment.type === 'change') {
     });
-    scope.applySyncFragments(syncFragments, options, function(err, results) {
+    scope.applySyncFragmentsWithLock(syncFragments, options, function(err, results) {
         if (err) {
@@ -297,0 +297,0 @@ return callback(err);

lib/query/pull_query_operation.js

@@ -200,3 +200,11 @@ // Copyright (c) 2015 Uber Technologies, Inc.
     // Bind all the late bound sync fragments and then apply them
+    var writeLockRelease;
     async.waterfall([
+        function aquireScopeWriteLock(nextCallback) {
+            scope.lock.writeLock(function(release) {
+                writeLockRelease = release;
+                nextCallback();
+            });
+        },
+
         function bindlateBoundSyncFragments(nextCallback) {
@@ -254,20 +262,12 @@ var changeSyncFragments = this.lateBoundSyncFragments;
 
-        function applySyncFragments(nextCallback) {
-            var syncFragments = this.lateBoundSyncFragments.filter(function(syncFragment) {
-                return Object.keys(syncFragment.properties).length > 0;
-            });
+        function applySyncFragmentsAndCreateQueryResult(nextCallback) {
+            var fragments = this.lateBoundSyncFragments;
             var options = this.options;
-            scope.applySyncFragments(syncFragments, options, function(err, results) {
-                nextCallback(err, syncFragments, results);
-            });
+            this._applySyncFragmentsWithLockAndCreateQueryResult(scope, fragments, options, modelObject, nextCallback);
         }.bind(this),
 
-        function createQueryResult(syncFragments, results, nextCallback) {
-            var result = this._createQueryResult(modelObject, syncFragments, results);
-            nextCallback(null, result);
-        }.bind(this)
-
     ], function(err, result) {
         this.executed = true;
         this.executing = false;
+        writeLockRelease();
         callback(err, result);
@@ -274,0 +274,0 @@ }.bind(this));

lib/query/push_query_operation.js

@@ -154,3 +154,11 @@ // Copyright (c) 2015 Uber Technologies, Inc.
     // Bind all the late bound sync fragments and then apply them
+    var writeLockRelease;
     async.waterfall([
+        function aquireScopeWriteLock(nextCallback) {
+            scope.lock.writeLock(function(release) {
+                writeLockRelease = release;
+                nextCallback();
+            });
+        },
+
         function bindLateBoundChangeSyncFragments(nextCallback) {
@@ -210,3 +218,3 @@ var changeSyncFragments = _.filter(this.lateBoundSyncFragments, function(syncFragment) {
             var options = this.options;
-            this._applySyncFragmentsAndCreateQueryResult(scope, fragments, options, modelObject, nextCallback);
+            this._applySyncFragmentsWithLockAndCreateQueryResult(scope, fragments, options, modelObject, nextCallback);
         }.bind(this),
@@ -217,2 +225,3 @@
         this.executing = false;
+        writeLockRelease();
         callback(err, result);
@@ -219,0 +228,0 @@ }.bind(this));

lib/query/set_query_operation.js

@@ -140,3 +140,11 @@ // Copyright (c) 2015 Uber Technologies, Inc.
     // Bind all the late bound sync fragments and then apply them
+    var writeLockRelease;
     async.waterfall([
+        function aquireScopeWriteLock(nextCallback) {
+            scope.lock.writeLock(function(release) {
+                writeLockRelease = release;
+                nextCallback();
+            });
+        },
+
         function bindLateBoundChangeSyncFragments(nextCallback) {
@@ -173,3 +181,3 @@ var changeSyncFragments = _.filter(this.lateBoundSyncFragments, function(syncFragment) {
             var options = this.options;
-            this._applySyncFragmentsAndCreateQueryResult(scope, fragments, options, modelObject, nextCallback);
+            this._applySyncFragmentsWithLockAndCreateQueryResult(scope, fragments, options, modelObject, nextCallback);
         }.bind(this),
@@ -180,2 +188,3 @@
         this.executing = false;
+        writeLockRelease();
         callback(err, result);
@@ -182,0 +191,0 @@ }.bind(this));

lib/scope.js

@@ -33,2 +33,3 @@ // Copyright (c) 2015 Uber Technologies, Inc.
 var Query = require('./query/query');
+var ReadWriteLock = require('rwlock');
 var StorageMiddleware = require('./middleware/storage/storage_middleware');
@@ -51,2 +52,3 @@ var SyncFragment = require('./sync_fragment');
     this.params = options.params || {};
+    this.lock = new ReadWriteLock();
 
@@ -62,4 +64,2 @@ if (typeof options.disableProcedureConstraints === 'boolean') {
     this._rootModelObjectConstructor = null;
-    this._applyingSyncFragments = false;
-    this._applySyncFragmentsQueue = [];
     this._orphanedModelObjects = {};
@@ -240,8 +240,15 @@ this._onModelObjectParentAddedListener = this._onModelObjectParentAdded.bind(this);
 Scope.prototype.applySyncFragments = function(syncFragments, options, callback) {
-    // Queue this call if currently applying a set of sync fragments
-    if (this._applyingSyncFragments) {
-        return this._applySyncFragmentsQueue.push(arguments);
-    }
-    this._applyingSyncFragments = true;
+    this.lock.writeLock(function(release) {
+        this.applySyncFragmentsWithLock(syncFragments, options, function(err, results) {
+            release();
+            if (err) {
+                callbackOrEmitError(this, callback, err);
+            } else {
+                maybeCallback(callback)(null, results);
+            }
+        }.bind(this));
+    }.bind(this));
+};
 
+Scope.prototype.applySyncFragmentsWithLock = function(syncFragments, options, callback) {
     var optionsType = typeof options;
@@ -454,11 +461,2 @@ if (optionsType === 'function') {
     ], function(err, results) {
-        // If any pending calls were queued then process the next on the next run loop
-        this._applyingSyncFragments = false;
-        if (this._applySyncFragmentsQueue.length > 0) {
-            var nextArguments = this._applySyncFragmentsQueue.pop();
-            process.nextTick(function() {
-                this.applySyncFragments.apply(this, nextArguments);
-            }.bind(this));
-        }
-
         if (err) {
@@ -465,0 +463,0 @@ // TODO: discard this scope, integrity is GOOOOONE

package.json

 {
   "name": "jetstream",
-  "version": "0.10.0",
+  "version": "0.11.0",
   "description": "Jetstream Sync server framework to sync local and remote models",
@@ -54,2 +54,3 @@ "keywords": [
     "rust-result": "^0.1.0",
+    "rwlock": "^5.0.0",
     "semver": "^4.0.0",
@@ -56,0 +57,0 @@ "tryit": "^1.0.1",

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

507598

increased by0.06%

Lines of code

11365

increased by0.13%

Worsened metrics

Dependency count

21

increased by5%

Dependency changes

• + Addedrwlock@^5.0.0

• + Addedrwlock@5.0.0(transitive)