Merge 'upstream/master' into fix-hasWritePending-in-op-callback

Author: gkubisa

.gitignore

@@ -2,6 +2,9 @@
 *.swp
 *.DS_Store
 
+# WebStorm
+.idea/
+
 # Emacs
 \#*\#
 

.travis.yml

@@ -1,9 +1,9 @@
 language: node_js
 node_js:
+  - "10"
   - "9"
   - "8"
   - "6"
-  - "4"
 script: "npm run jshint && npm run test-cover"
 # Send coverage data to Coveralls
 after_script: "cat ./coverage/lcov.info | ./node_modules/coveralls/bin/coveralls.js"

README.md

@@ -123,14 +123,15 @@ Register a new middleware.
   One of:
   * `'connect'`: A new client connected to the server.
   * `'op'`: An operation was loaded from the database.
-  * `'doc'`: A snapshot was loaded from the database.
+  * `'doc'`: DEPRECATED: A snapshot was loaded from the database. Please use 'readSnapshots'
+  * `'readSnapshots'`: Snapshot(s) were loaded from the database for a fetch or subscribe of a query or document
   * `'query'`: A query is about to be sent to the database
-  * `'submit'`: An operation is about to be submited to the database
+  * `'submit'`: An operation is about to be submitted to the database
   * `'apply'`: An operation is about to be applied to a snapshot
     before being committed to the database
   * `'commit'`: An operation was applied to a snapshot; The operation
     and new snapshot are about to be written to the database.
-  * `'after submit'`: An operation was successfully submitted to
+  * `'afterSubmit'`: An operation was successfully submitted to
     the database.
   * `'receive'`: Received a message from a client
 * `fn` _(Function(request, callback))_
@@ -141,6 +142,7 @@ Register a new middleware.
   * `req`: The HTTP request being handled
   * `collection`: The collection name being handled
   * `id`: The document id being handled
+  * `snapshots`: The retrieved snapshots for the `readSnapshots` action
   * `query`: The query object being handled
   * `op`: The op being handled
 

lib/backend.js

@@ -32,10 +32,63 @@ function Backend(options) {
   // The number of open agents for monitoring and testing memory leaks
   this.agentsCount = 0;
   this.remoteAgentsCount = 0;
+
+  // The below shims are for backwards compatibility. These options will be
+  // removed in a future major version
+  if (!options.disableDocAction) {
+    this._shimDocAction();
+  }
+  if (!options.disableSpaceDelimitedActions) {
+    this._shimAfterSubmit();
+  }
 }
 module.exports = Backend;
 emitter.mixin(Backend);
 
+Backend.prototype.MIDDLEWARE_ACTIONS = {
+  // An operation was successfully submitted to the database.
+  afterSubmit: 'afterSubmit',
+  // DEPRECATED: Synonym for 'afterSubmit'
+  'after submit': 'after submit',
+  // An operation is about to be applied to a snapshot before being committed to the database
+  apply: 'apply',
+  // An operation was applied to a snapshot; The operation and new snapshot are about to be written to the database.
+  commit: 'commit',
+  // A new client connected to the server.
+  connect: 'connect',
+  // DEPRECATED: A snapshot was loaded from the database.
+  doc: 'doc',
+  // An operation was loaded from the database
+  op: 'op',
+  // A query is about to be sent to the database
+  query: 'query',
+  // Received a message from a client
+  receive: 'receive',
+  // Snapshot(s) were received from the database and are about to be returned to a client
+  readSnapshots: 'readSnapshots',
+  // An operation is about to be submitted to the database
+  submit: 'submit'
+};
+
+Backend.prototype._shimDocAction = function() {
+  var backend = this;
+  this.use(this.MIDDLEWARE_ACTIONS.readSnapshots, function(request, callback) {
+    async.each(request.snapshots, function(snapshot, eachCb) {
+      var docRequest = {collection: request.collection, id: snapshot.id, snapshot: snapshot};
+      backend.trigger(backend.MIDDLEWARE_ACTIONS.doc, request.agent, docRequest, eachCb);
+    }, callback);
+  });
+};
+
+// Shim for backwards compatibility with deprecated middleware action name.
+// The action 'after submit' is now 'afterSubmit'.
+Backend.prototype._shimAfterSubmit = function() {
+  var backend = this;
+  this.use(backend.MIDDLEWARE_ACTIONS.afterSubmit, function(request, callback) {
+    backend.trigger(backend.MIDDLEWARE_ACTIONS['after submit'], request.agent, request, callback);
+  });
+};
+
 Backend.prototype.close = function(callback) {
   var wait = 3;
   var backend = this;
@@ -82,7 +135,7 @@ Backend.prototype.connect = function(connection, req) {
  */
 Backend.prototype.listen = function(stream, req) {
   var agent = new Agent(this, stream);
-  this.trigger('connect', agent, {stream: stream, req: req}, function(err) {
+  this.trigger(this.MIDDLEWARE_ACTIONS.connect, agent, {stream: stream, req: req}, function(err) {
     if (err) return agent.close(err);
     agent._open();
   });
@@ -155,11 +208,11 @@ Backend.prototype.submit = function(agent, index, id, op, options, callback) {
   if (err) return callback(err);
   var request = new SubmitRequest(this, agent, index, id, op, options);
   var backend = this;
-  backend.trigger('submit', agent, request, function(err) {
+  backend.trigger(backend.MIDDLEWARE_ACTIONS.submit, agent, request, function(err) {
     if (err) return callback(err);
     request.submit(function(err) {
       if (err) return callback(err);
-      backend.trigger('after submit', agent, request, function(err) {
+      backend.trigger(backend.MIDDLEWARE_ACTIONS.afterSubmit, agent, request, function(err) {
         if (err) return callback(err);
         backend._sanitizeOps(agent, request.projection, request.collection, id, request.ops, function(err) {
           if (err) return callback(err);
@@ -179,7 +232,7 @@ Backend.prototype._sanitizeOp = function(agent, projection, collection, id, op,
       return callback(err);
     }
   }
-  this.trigger('op', agent, {collection: collection, id: id, op: op}, callback);
+  this.trigger(this.MIDDLEWARE_ACTIONS.op, agent, {collection: collection, id: id, op: op}, callback);
 };
 Backend.prototype._sanitizeOps = function(agent, projection, collection, id, ops, callback) {
   var backend = this;
@@ -194,33 +247,31 @@ Backend.prototype._sanitizeOpsBulk = function(agent, projection, collection, ops
   }, callback);
 };
 
-Backend.prototype._sanitizeSnapshot = function(agent, projection, collection, id, snapshot, callback) {
+Backend.prototype._sanitizeSnapshots = function(agent, projection, collection, snapshots, callback) {
   if (projection) {
     try {
-      projections.projectSnapshot(projection.fields, snapshot);
+      projections.projectSnapshots(projection.fields, snapshots);
     } catch (err) {
       return callback(err);
     }
   }
-  this.trigger('doc', agent, {collection: collection, id: id, snapshot: snapshot}, callback);
-};
-Backend.prototype._sanitizeSnapshots = function(agent, projection, collection, snapshots, callback) {
-  var backend = this;
-  async.each(snapshots, function(snapshot, eachCb) {
-    backend._sanitizeSnapshot(agent, projection, collection, snapshot.id, snapshot, eachCb);
-  }, callback);
-};
-Backend.prototype._sanitizeSnapshotBulk = function(agent, projection, collection, snapshotMap, callback) {
-  var backend = this;
-  async.forEachOf(snapshotMap, function(snapshot, id, eachCb) {
-    backend._sanitizeSnapshot(agent, projection, collection, id, snapshot, eachCb);
-  }, callback);
+  var request = {collection: collection, snapshots: snapshots};
+  this.trigger(this.MIDDLEWARE_ACTIONS.readSnapshots, agent, request, callback);
 };
 
 Backend.prototype._getSnapshotProjection = function(db, projection) {
   return (db.projectsSnapshots) ? null : projection;
 };
 
+Backend.prototype._getSnapshotsFromMap = function(ids, snapshotMap) {
+  var snapshots = new Array(ids.length);
+  for (var i = 0; i < ids.length; i++) {
+    var id = ids[i];
+    snapshots[i] = snapshotMap[id];
+  }
+  return snapshots;
+};
+
 // Non inclusive - gets ops from [from, to). Ie, all relevant ops. If to is
 // not defined (null or undefined) then it returns all ops.
 Backend.prototype.getOps = function(agent, index, id, from, to, callback) {
@@ -283,7 +334,8 @@ Backend.prototype.fetch = function(agent, index, id, callback) {
   backend.db.getSnapshot(collection, id, fields, null, function(err, snapshot) {
     if (err) return callback(err);
     var snapshotProjection = backend._getSnapshotProjection(backend.db, projection);
-    backend._sanitizeSnapshot(agent, snapshotProjection, collection, id, snapshot, function(err) {
+    var snapshots = [snapshot];
+    backend._sanitizeSnapshots(agent, snapshotProjection, collection, snapshots, function(err) {
       if (err) return callback(err);
       backend.emit('timing', 'fetch', Date.now() - start, request);
       callback(null, snapshot);
@@ -306,7 +358,8 @@ Backend.prototype.fetchBulk = function(agent, index, ids, callback) {
   backend.db.getSnapshotBulk(collection, ids, fields, null, function(err, snapshotMap) {
     if (err) return callback(err);
     var snapshotProjection = backend._getSnapshotProjection(backend.db, projection);
-    backend._sanitizeSnapshotBulk(agent, snapshotProjection, collection, snapshotMap, function(err) {
+    var snapshots = backend._getSnapshotsFromMap(ids, snapshotMap);
+    backend._sanitizeSnapshots(agent, snapshotProjection, collection, snapshots, function(err) {
       if (err) return callback(err);
       backend.emit('timing', 'fetchBulk', Date.now() - start, request);
       callback(null, snapshotMap);
@@ -479,7 +532,7 @@ Backend.prototype._triggerQuery = function(agent, index, query, options, callbac
     snapshotProjection: null,
   };
   var backend = this;
-  backend.trigger('query', agent, request, function(err) {
+  backend.trigger(backend.MIDDLEWARE_ACTIONS.query, agent, request, function(err) {
     if (err) return callback(err);
     // Set the DB reference for the request after the middleware trigger so
     // that the db option can be changed in middleware

lib/projections.js

@@ -1,6 +1,7 @@
 var json0 = require('ot-json0').type;
 
 exports.projectSnapshot = projectSnapshot;
+exports.projectSnapshots = projectSnapshots;
 exports.projectOp = projectOp;
 exports.isSnapshotAllowed = isSnapshotAllowed;
 exports.isOpAllowed = isOpAllowed;
@@ -15,6 +16,13 @@ function projectSnapshot(fields, snapshot) {
   snapshot.data = projectData(fields, snapshot.data);
 }
 
+function projectSnapshots(fields, snapshots) {
+  for (var i = 0; i < snapshots.length; i++) {
+    var snapshot = snapshots[i];
+    projectSnapshot(fields, snapshot);
+  }
+}
+
 function projectOp(fields, op) {
   if (op.create) {
     projectSnapshot(fields, op.create);

lib/pubsub/index.js

@@ -1,8 +1,12 @@
+var emitter = require('../emitter');
 var OpStream = require('../op-stream');
 var ShareDBError = require('../error');
 var util = require('../util');
 
 function PubSub(options) {
+  if (!(this instanceof PubSub)) return new PubSub(options);
+  emitter.EventEmitter.call(this);
+
   this.prefix = options && options.prefix;
   this.nextStreamId = 1;
   this.streamsCount = 0;
@@ -13,8 +17,14 @@ function PubSub(options) {
   // isn't complete until the callback returns from Redis
   // Maps channel -> true
   this.subscribed = {};
+
+  var pubsub = this;
+  this._defaultCallback = function(err) {
+    if (err) return pubsub.emit('error', err);
+  };
 }
 module.exports = PubSub;
+emitter.mixin(PubSub);
 
 PubSub.prototype.close = function(callback) {
   for (var channel in this.streams) {
@@ -23,22 +33,29 @@ PubSub.prototype.close = function(callback) {
       map[id].destroy();
     }
   }
-  if (callback) callback();
+  if (callback) process.nextTick(callback);
 };
 
 PubSub.prototype._subscribe = function(channel, callback) {
-  callback(new ShareDBError(5015, '_subscribe PubSub method unimplemented'));
+  process.nextTick(function() {
+    callback(new ShareDBError(5015, '_subscribe PubSub method unimplemented'));
+  });
 };
 
 PubSub.prototype._unsubscribe = function(channel, callback) {
-  callback(new ShareDBError(5016, '_unsubscribe PubSub method unimplemented'));
+  process.nextTick(function() {
+    callback(new ShareDBError(5016, '_unsubscribe PubSub method unimplemented'));
+  });
 };
 
 PubSub.prototype._publish = function(channels, data, callback) {
-  callback(new ShareDBError(5017, '_publish PubSub method unimplemented'));
+  process.nextTick(function() {
+    callback(new ShareDBError(5017, '_publish PubSub method unimplemented'));
+  });
 };
 
 PubSub.prototype.subscribe = function(channel, callback) {
+  if (!callback) callback = this._defaultCallback;
   if (this.prefix) {
     channel = this.prefix + ' ' + channel;
   }
@@ -60,6 +77,7 @@ PubSub.prototype.subscribe = function(channel, callback) {
 };
 
 PubSub.prototype.publish = function(channels, data, callback) {
+  if (!callback) callback = this._defaultCallback;
   if (this.prefix) {
     for (var i = 0; i < channels.length; i++) {
       channels[i] = this.prefix + ' ' + channels[i];
@@ -109,9 +127,7 @@ PubSub.prototype._removeStream = function(channel, stream) {
   // complete before we can count on being subscribed again
   delete this.subscribed[channel];
 
-  this._unsubscribe(channel, function(err) {
-    if (err) throw err;
-  });
+  this._unsubscribe(channel, this._defaultCallback);
 };
 
 function shallowCopy(object) {

lib/query-emitter.js

@@ -185,7 +185,8 @@ QueryEmitter.prototype.queryPoll = function(callback) {
       if (inserted.length) {
         emitter.db.getSnapshotBulk(emitter.collection, inserted, emitter.fields, null, function(err, snapshotMap) {
           if (err) return emitter._finishPoll(err, callback, pending);
-          emitter.backend._sanitizeSnapshotBulk(emitter.agent, emitter.snapshotProjection, emitter.collection, snapshotMap, function(err) {
+          var snapshots = emitter.backend._getSnapshotsFromMap(inserted, snapshotMap);
+          emitter.backend._sanitizeSnapshots(emitter.agent, emitter.snapshotProjection, emitter.collection, snapshots, function(err) {
             if (err) return emitter._finishPoll(err, callback, pending);
             emitter._emitTiming('queryEmitter.pollGetSnapshotBulk', start);
             var diff = mapDiff(idsDiff, snapshotMap);
@@ -232,10 +233,10 @@ QueryEmitter.prototype.queryPollDoc = function(id, callback) {
       // delay in sending to the client anyway
       emitter.db.getSnapshot(emitter.collection, id, emitter.fields, null, function(err, snapshot) {
         if (err) return callback(err);
-        emitter.backend._sanitizeSnapshot(emitter.agent, emitter.snapshotProjection, emitter.collection, id, snapshot, function(err) {
+        var snapshots = [snapshot];
+        emitter.backend._sanitizeSnapshots(emitter.agent, emitter.snapshotProjection, emitter.collection, snapshots, function(err) {
           if (err) return callback(err);
-          var values = [snapshot];
-          emitter.onDiff([new arraydiff.InsertDiff(index, values)]);
+          emitter.onDiff([new arraydiff.InsertDiff(index, snapshots)]);
           emitter._emitTiming('queryEmitter.pollDocGetSnapshot', start);
           callback();
         });

package.json

@@ -1,6 +1,6 @@
 {
   "name": "sharedb",
-  "version": "1.0.0-beta.8",
+  "version": "1.0.0-beta.9",
   "description": "JSON OT database backend",
   "main": "lib/index.js",
   "dependencies": {