added support for the scale with meteor cluster

added supporting functions and hooks to meteor cluster to get simply plugged in

Author: arunoda

laika/tests/scaling_support.js

@@ -0,0 +1,53 @@
+var assert = require('assert');
+require('./common');
+
+suite('Scaling Support', function() {
+ test('trigger a client using .emitToSubscriptions()', function(done, server, client) {
+ server.evalSync(createServerStream, 'hello');
+ client.evalSync(createClientStream, 'hello');
+
+ client.evalSync(function() {
+ helloStream.on('evt', function(data) {
+ emit('evt', data, this);
+ });
+ emit('return');
+ });
+
+ client.on('evt', function(data, context) {
+ assert.deepEqual(data, {a: 10});
+ assert.deepEqual(context, {subscriptionId: 'subscriptionId', userId: 'userId'});
+ done();
+ });
+
+ server.evalSync(function() {
+ helloStream.emitToSubscriptions(['evt', {a: 10}], 'subscriptionId', 'userId');
+ });
+ });
+
+ test('listen to all events with firehose', function(done, server, client) {
+ server.evalSync(createServerStream, 'hello');
+ client.evalSync(createClientStream, 'hello');
+
+ client.evalSync(laika.actions.createUser, {username: 'arunoda', password: 'password'});
+ var user = client.evalSync(laika.actions.loggedInUser);
+
+ server.evalSync(function() {
+ helloStream.firehose = function(args, subscriptionId, userId) {
+ emit('firehose', args, subscriptionId, userId);
+ };
+ emit('return');
+ });
+
+ server.on('firehose', function(args, subscriptionId, userId) {
+ assert.deepEqual(args, ['evt', {a: 100}]);
+ assert.equal(userId, user._id);
+ assert.ok(subscriptionId);
+ done();
+ }); 
+
+ client.evalSync(function() {
+ helloStream.emit('evt', {a: 100});
+ emit('return');
+ });
+ });
+});

lib/server.js

@@ -4,6 +4,7 @@ var Fibers = Npm.require('fibers');
 
 Meteor.Stream = function Stream(name) {
  EV.call(this);
+
  var self = this;
  var streamName = 'stream-' + name;
  var allowFunction;
@@ -18,22 +19,18 @@ Meteor.Stream = function Stream(name) {
 
  self._emit = self.emit;
  self.emit = function emit() {
- var args = [];
- for(var key in arguments) {
- args.push(arguments[key]);
- }
- emitToSubscriptions(args, null, null);
+ self.emitToSubscriptions(arguments, null, null);
  };
 
  self.permissions = new Meteor.Stream.Permission(Meteor.Collection.insecure, true);
 
- self.addFilter = function(callback) {
+ self.addFilter = function addFilter(callback) {
  filters.push(callback);
  };
 
- function emitToSubscriptions(args, subscriptionId, userId) {
+ self.emitToSubscriptions = function emitToSubscriptions(args, subscriptionId, userId) {
  events.emit('item', {args: args, userId: userId, subscriptionId: subscriptionId});
- }
+ };
 
  Meteor.publish(streamName, function() {
  var subscriptionId = Random.id();
@@ -81,7 +78,11 @@ Meteor.Stream = function Stream(name) {
  if(methodContext.allowed) {
  //apply filters
  args = applyFilters(args, methodContext);
- emitToSubscriptions(args, subscriptionId, methodContext.userId);
+ self.emitToSubscriptions(args, subscriptionId, methodContext.userId);
+ //send to firehose if exists
+ if(self.firehose) {
+ self.firehose(args, subscriptionId, methodContext.userId);
+ }
  }
  //need to send this to server always
  self._emit.apply(methodContext, args);
@@ -105,46 +106,4 @@ Meteor.Stream = function Stream(name) {
  }
 };
 
-util.inherits(Meteor.Stream, EV);
-
-Meteor.Stream.Permission = function (acceptAll, cacheAll) {
-
- var options = {
- "read": {
- results: {}
- }, 
- "write": {
- results: {}
- }
- };
-
- this.read = function(func, cache) {
- options['read']['func'] = func;
- options['read']['doCache'] = (cache === undefined)? cacheAll: cache; 
- };
-
- this.write = function(func, cache) {
- options['write']['func'] = func;
- options['write']['doCache'] = (cache === undefined)? cacheAll: cache; 
- };
-
- this.checkPermission = function(type, userId, eventName) {
- var namespace = userId + '-' + eventName;
- var result = options[type].results[namespace];
- 
- if(result === undefined) {
- var func = options[type].func;
- if(func) {
- result = func(userId, eventName);
- if(options[type].doCache) {
- options[type].results[namespace] = result;
- return result;
- }
- } else {
- return acceptAll;
- }
- } else {
- return result;
- }
- }; 
-}
+util.inherits(Meteor.Stream, EV);

lib/stream_permission.js

@@ -0,0 +1,40 @@
+Meteor.Stream.Permission = function (acceptAll, cacheAll) {
+ var options = {
+ "read": {
+ results: {}
+ }, 
+ "write": {
+ results: {}
+ }
+ };
+
+ this.read = function(func, cache) {
+ options['read']['func'] = func;
+ options['read']['doCache'] = (cache === undefined)? cacheAll: cache; 
+ };
+
+ this.write = function(func, cache) {
+ options['write']['func'] = func;
+ options['write']['doCache'] = (cache === undefined)? cacheAll: cache; 
+ };
+
+ this.checkPermission = function(type, userId, eventName) {
+ var namespace = userId + '-' + eventName;
+ var result = options[type].results[namespace];
+ 
+ if(result === undefined) {
+ var func = options[type].func;
+ if(func) {
+ result = func(userId, eventName);
+ if(options[type].doCache) {
+ options[type].results[namespace] = result;
+ return result;
+ }
+ } else {
+ return acceptAll;
+ }
+ } else {
+ return result;
+ }
+ }; 
+}

package.js

@@ -3,6 +3,6 @@ Package.describe({
 });
 
 Package.on_use(function (api, where) {
- api.add_files(['lib/ev.js', 'lib/server.js'], 'server');
+ api.add_files(['lib/ev.js', 'lib/server.js', 'lib/stream_permission.js'], 'server');
  api.add_files(['lib/ev.js', '/lib/client.js'], 'client');
 });

test_cases.todo

@@ -4,10 +4,8 @@ Without Permissions:
  ✔ Client to Server @done (13-06-20 15:52)
  ✔ Client to Client @done (13-06-20 15:52)
  ✔ Client to Client with logged in @done (13-06-21 16:31)
- ☐ Server Server
  ✔ emitting not fired to me in the client @done (13-06-22 17:52)
 
-
 Read Permissions:
 
  ✔ Server to Client with denied @done (13-06-21 14:55)
@@ -34,7 +32,7 @@ Filters:
  ✔ simple filter adding an arg @done (13-06-23 09:04)
  ✔ multiple filters @done (13-06-23 09:06)
  ✔ modifying an arg @done (13-06-23 09:08)
- ☐ filters with userId and subscriptionId
+ ✔ filters with userId and subscriptionId @done (13-06-26 11:10)
 
 onDisconnect Handlers:
  ✔ single onDisconnect handler @done (13-06-24 14:01)
@@ -50,6 +48,9 @@ EV:
  ✔ once and emit @done (13-06-24 15:04)
  ✔ emit with different contexts @done (13-06-24 16:06)
 
+Scaling Support:
+ ✔ firehose @done (13-06-26 15:08)
+ ✔ emitToSubscribers @done (13-06-26 15:08)