Merge pull request #725 from RocketChat/stream-notify

Stream notify

client/lib/RoomManager.coffee

@@ -30,12 +30,15 @@ Meteor.startup ->
 				ChatMessage.update {_id: recordAfter._id}, {$set: {tick: new Date}}
 
 
+onDeleteMessageStream = (msg) ->
+	ChatMessage.remove _id: msg._id
+
+
 @RoomManager = new class
 	defaultTime = 600000 # 10 minutes
 	openedRooms = {}
 	subscription = null
 	msgStream = new Meteor.Stream 'messages'
-	deleteMsgStream = new Meteor.Stream 'delete-message'
 	onlineUsers = new ReactiveVar {}
 
 	Dep = new Tracker.Dependency
@@ -52,7 +55,7 @@ Meteor.startup ->
 
 			if openedRooms[typeName].rid?
 				msgStream.removeListener openedRooms[typeName].rid
-				deleteMsgStream.removeListener openedRooms[typeName].rid
+				RocketChat.Notifications.unRoom openedRooms[typeName].rid, 'deleteMessage', onDeleteMessageStream
 
 			openedRooms[typeName].ready = false
 			openedRooms[typeName].active = false
@@ -101,8 +104,7 @@ Meteor.startup ->
 										RoomManager.close type + FlowRouter.getParam('name')
 										FlowRouter.go FlowRouter.current().route.name, name: msg.msg
 
-						deleteMsgStream.on openedRooms[typeName].rid, (msg) ->
-							ChatMessage.remove _id: msg._id
+						RocketChat.Notifications.onRoom openedRooms[typeName].rid, 'deleteMessage', onDeleteMessageStream
 
 				Dep.changed()
 

client/lib/msgTyping.coffee

@@ -1,5 +1,4 @@
 @MsgTyping = do ->
-	stream = new Meteor.Stream 'typing'
 	timeout = 15000
 	timeouts = {}
 	renew = true
@@ -11,7 +10,7 @@
 	addStream = (room) ->
 		if _.isEmpty usersTyping[room]?.users
 			usersTyping[room] = { users: {} }
-			stream.on room, (typing) ->
+			RocketChat.Notifications.onRoom room, 'typing', (typing) ->
 				unless typing?.username is Meteor.user()?.username
 					if typing.start
 						users = usersTyping[room].users
@@ -41,7 +40,7 @@
 
 		renew = false
 		selfTyping.set true
-		stream.emit 'typing', { room: room, username: Meteor.user()?.username, start: true }
+		RocketChat.Notifications.notifyRoom room, 'typing', { room: room, username: Meteor.user()?.username, start: true }
 		clearTimeout timeouts[room]
 		timeouts[room] = Meteor.setTimeout ->
 			stop(room)
@@ -51,20 +50,20 @@
 		renew = true
 		selfTyping.set false
 		if timeouts?[room]?
-			clearTimeout(timeouts[room]) 
+			clearTimeout(timeouts[room])
 			timeouts[room] = null
-		stream.emit 'typing', { room: room, username: Meteor.user()?.username, stop: true }
-		
+		RocketChat.Notifications.notifyRoom room, 'typing', { room: room, username: Meteor.user()?.username, stop: true }
+
 	get = (room) ->
 		dep.depend()
 		unless usersTyping[room]
 			usersTyping[room] = { users: {} }
 		users = usersTyping[room].users
 		return _.keys(users) or []
 
-	return { 
+	return {
 		start: start
 		stop: stop
 		get: get
 		selfTyping: selfTyping
-	}
+	}

client/notifications/updateAvatar.coffee

@@ -0,0 +1,3 @@
+Meteor.startup ->
+	RocketChat.Notifications.onAll 'updateAvatar', (data) ->
+		updateAvatarOfUsername data.username

client/views/account/avatar/prompt.coffee

@@ -34,11 +34,9 @@ Template.avatarPrompt.events
 	'click .select-service': ->
 		if @service is 'initials'
 			Meteor.call 'resetAvatar'
-			updateAvatarOfUsername Meteor.user().username
 			toastr.success t('Avatar_changed_successfully')
 		else
 			Meteor.call 'setAvatarFromService', @blob, @contentType, @service, ->
-				updateAvatarOfUsername Meteor.user().username
 				toastr.success t('Avatar_changed_successfully')
 
 	'click .login-with-service': (event, template) ->

packages/meteor-streams/LICENSE

@@ -0,0 +1,22 @@
+(The MIT License)
+
+Copyright (c) 2013 Arunoda Susiripala <[email protected]>
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+'Software'), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
+CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
+TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
+SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

packages/meteor-streams/README.md

@@ -0,0 +1,10 @@
+# Meteor Streams [![Build Status](https://travis-ci.org/arunoda/meteor-streams.png?branch=master)](https://travis-ci.org/arunoda/meteor-streams)
+
+DB less realtime communication for meteor
+
+## Development Status
+Project development status is [inactive](https://github.com/arunoda/meteor-streams/issues/21#issuecomment-59030380).
+
+## [Documentation](http://arunoda.github.io/meteor-streams/)
+
+[![Meteor Streams - DB less realtime communication for meteor](http://i.imgur.com/ZB3g3AK.png)](http://arunoda.github.io/meteor-streams/)

packages/meteor-streams/lib/client.js

@@ -0,0 +1,48 @@
+Meteor.Stream = function Stream(name, callback) {
+  EV.call(this);
+
+  var self = this;
+  var streamName = 'stream-' + name;
+  var collection = new Meteor.Collection(streamName);
+  var subscription;
+  var subscriptionId;
+
+  var connected = false;
+  var pendingEvents = [];
+
+  self._emit = self.emit;
+
+  collection.find({}).observe({
+    "added": function(item) {
+      if(item.type == 'subscriptionId') {
+        subscriptionId = item._id;
+        connected = true;
+        pendingEvents.forEach(function(args) {
+          self.emit.apply(self, args);
+        });
+        pendingEvents = [];
+      } else {
+        var context = {};
+        context.subscriptionId = item.subscriptionId;
+        context.userId = item.userId;
+        self._emit.apply(context, item.args);    
+      }
+    }
+  });
+
+  subscription = Meteor.subscribe(streamName, callback);
+
+  self.emit = function emit() {
+    if(connected) {
+      Meteor.call(streamName, subscriptionId, arguments);
+    } else {
+      pendingEvents.push(arguments);
+    }
+  };
+
+  self.close = function close() {
+    subscription.stop();
+  };
+}
+
+_.extend(Meteor.Stream.prototype, EV.prototype);

packages/meteor-streams/lib/ev.js

@@ -0,0 +1,44 @@
+function _EV() {
+  var self = this;
+  var handlers = {};
+
+  self.emit = function emit(event) {
+    var args = Array.prototype.slice.call(arguments, 1);
+
+    if(handlers[event]) {
+      for(var lc=0; lc<handlers[event].length; lc++) {
+        var handler = handlers[event][lc];
+        handler.apply(this, args);
+      }
+    }
+  };
+
+  self.on = function on(event, callback) {
+    if(!handlers[event]) {
+      handlers[event] = [];
+    }
+    handlers[event].push(callback);
+  };
+
+  self.once = function once(event, callback) {
+    self.on(event, function onetimeCallback() {
+      callback.apply(this, arguments);
+      self.removeListener(event, onetimeCallback);
+    });
+  };
+
+  self.removeListener = function removeListener(event, callback) {
+    if(handlers[event]) {
+      var index = handlers[event].indexOf(callback);
+      console.log(index);
+      if (index > -1)
+        handlers[event].splice(index, 1);
+    }
+  };
+
+  self.removeAllListeners = function removeAllListeners(event) {
+    handlers[event] = undefined;
+  };
+}
+
+EV = _EV;

packages/meteor-streams/lib/server.js

@@ -0,0 +1,114 @@
+var EventEmitter = Npm.require('events').EventEmitter;
+var util = Npm.require('util');
+var Fibers = Npm.require('fibers');
+
+Meteor.Stream = function Stream(name) {
+  EV.call(this);
+
+  var self = this;
+  var streamName = 'stream-' + name;
+  var allowFunction;
+  var allowResultCache = true;
+  var allowResults = {};
+  var filters = [];
+
+  self.name = name;
+
+  var events = new EventEmitter();
+  events.setMaxListeners(0);
+
+  var disconnectEvents = new EV();
+
+  self._emit = self.emit;
+  self.emit = function emit() {
+    self.emitToSubscriptions(arguments, null, null);
+  };
+
+  var defaultResult =  (typeof(Package) == 'object' && Package.insecure)? true: Meteor.Collection.insecure === true;
+  self.permissions = new Meteor.Stream.Permission(defaultResult, true);
+
+  self.addFilter = function addFilter(callback) {
+    filters.push(callback);
+  };
+
+  self.emitToSubscriptions = function emitToSubscriptions(args, subscriptionId, userId) {
+    events.emit('item', {args: args, userId: userId, subscriptionId: subscriptionId});
+  };
+
+  Meteor.publish(streamName, function() {
+    check(arguments, Match.Any);
+    var subscriptionId = Random.id();
+    var publication = this;
+
+    //send subscription id as the first document
+    publication.added(streamName, subscriptionId, {type: 'subscriptionId'});
+    publication.ready();
+    events.on('item', onItem);
+
+    function onItem(item) {
+      Fibers(function() {
+        var id = Random.id();
+        if(self.permissions.checkPermission('read', subscriptionId, publication.userId, item.args)) {
+          //do not send again this to the sender
+          if(subscriptionId != item.subscriptionId) {
+            publication.added(streamName, id, item);
+            publication.removed(streamName, id);
+          }
+        }
+      }).run();
+    }
+
+    publication.onStop(function() {
+      //trigger related onDisconnect handlers if exists
+      Fibers(function() {
+        disconnectEvents.emit(subscriptionId);
+        disconnectEvents.removeAllListeners(subscriptionId);
+      }).run();
+      events.removeListener('item', onItem);
+    });
+  });
+
+  var methods = {};
+  methods[streamName] = function(subscriptionId, args) {
+    check(arguments, Match.Any);
+    //in order to send this to the server callback
+    var userId = this.userId;
+    Fibers(function() {
+      var methodContext = {};
+      methodContext.userId = userId;
+      methodContext.subscriptionId = subscriptionId;
+
+      //in order to send this to the serve callback
+      methodContext.allowed = self.permissions.checkPermission('write', subscriptionId, methodContext.userId, args);
+      if(methodContext.allowed) {
+        //apply filters
+        args = applyFilters(args, methodContext);
+        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);
+
+      //register onDisconnect handlers if provided
+      if(typeof(methodContext.onDisconnect) == 'function') {
+        disconnectEvents.on(subscriptionId, methodContext.onDisconnect)
+      }
+
+    }).run();
+  };
+  Meteor.methods(methods);
+
+  function applyFilters(args, context) {
+    var eventName = args.shift();
+    filters.forEach(function(filter) {
+      args = filter.call(context, eventName, args);
+    });
+    args.unshift(eventName);
+    return args;
+  }
+};
+
+util.inherits(Meteor.Stream, EV);

packages/meteor-streams/lib/stream_permission.js

@@ -0,0 +1,42 @@
+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, subscriptionId, userId, args) {
+    var eventName = args[0];
+    var namespace = subscriptionId + '-' + eventName;
+    var result = options[type].results[namespace];
+
+    if(result === undefined) {
+      var func = options[type].func;
+      if(func) {
+        var context = {subscriptionId: subscriptionId, userId: userId};
+        result = func.apply(context, args);
+        if(options[type].doCache) {
+          options[type].results[namespace] = result;
+        }
+        return result;
+      } else {
+        return acceptAll;
+      }
+    } else {
+      return result;
+    }
+  };  
+}