Building a Real-Time Activity Stream on Cloud Foundry with Node.js, Redis and MongoDB--Part II

lib/socket-io-server.js

@@ -1,37 +1,136 @@
 module.exports = function Server(expressInstance, sessionStore) {
 	var parseCookie = require('connect').utils.parseCookie;
 	var io = require('socket.io').listen(expressInstance);
+    var asmsServer = expressInstance.asmsDB;
+    var thisApp = expressInstance.thisApp;
+    var thisInstance = expressInstance.thisInstance;
 
 	io.configure(function () {
 		io.set('log level', 0);
 	});
 
 	io.set('authorization', function(handshakeData, ack) {
 		var cookies = parseCookie(handshakeData.headers.cookie);
-		sessionStore.get(cookies['connect.sid'], function(err, sessionData) {
+		sessionStore.get(cookies[expressInstance.cookieName], function(err, sessionData) {
 			handshakeData.session = sessionData || {};
-			handshakeData.sid = cookies['connect.sid']|| null;
+			handshakeData.sid = cookies[expressInstance.cookieName]|| null;
 			ack(err, err ? false : true);
 		});
 	});
 
 	io.sockets.on('connection', function(client) {
 		var user = client.handshake.session.user ? client.handshake.session.user.name : 'UID: '+(client.handshake.session.uid || 'has no UID');
 
+        var desiredStream = "firehose";
+
+        if (client.handshake.session && client.handshake.session.desiredStream) {
+            desiredStream = client.handshake.session.desiredStream;
+        }
+
 		// Join user specific channel, this is good so content is send across user tabs.
 		client.join(client.handshake.sid);
 
-		client.send('welcome: '+user);
 
-		client.on('message', function(msg) {
-			// Send back the message to the users room.
-			io.sockets.in(client.handshake.sid).send('socket.io relay message "'+msg+'" from: '+ user +' @ '+new Date().toString().match(/[0-9]+:[0-9]+:[0-9]+/));
+        var avatarUrl = (client.handshake.session.auth && client.handshake.session.user && client.handshake.session.user.image) ? client.handshake.session.user.image : '/img/codercat-sm.jpg';
+        var currentUser = {displayName: user, image: {url: avatarUrl}};
+
+
+        console.log("Subscribing " + user);
+
+        asmsServer.subscribe(desiredStream,  function(channel, json) {
+            client.send(json);
+        });
+
+        var cf_provider;
+        var provider = new asmsServer.ActivityObject({'displayName': 'The Internet', icon: {url: ''}});
+        if (client.handshake.session.auth) {
+            if (client.handshake.session.auth.github) {
+                provider.displayName = 'GitHub';
+                provider.icon.url = 'http://github.com/favicon.ico';
+            } else if (client.handshake.session.auth.facebook) {
+                provider.displayName = 'Facebook';
+                provider.icon = {url: 'http://facebook.com/favicon.ico'};
+            } else if (client.handshake.session.auth.twitter) {
+                provider.displayName = 'Twitter';
+                provider.icon = {url: 'http://twitter.com/favicon.ico'};
+            }
+        }
+        provider.save(function(err) {
+            if (err == null) {
+                var cf_provider = new asmsServer.ActivityObject({'displayName': 'Cloud Foundry', icon:{url: 'http://www.cloudfoundry.com/images/favicon.ico'}});
+                   cf_provider.save(function(err) {
+                       if (err == null) {
+                           if (client.handshake.session && client.handshake.session.auth &&  client.handshake.session.user) {
+                               var act = new asmsServer.Activity({
+                                       id: 1,
+                                       actor: currentUser,
+                                       verb: 'connect',
+                                       object: thisInstance,
+                                       target: thisApp,
+                                       title: "connected to",
+                                       provider: provider,
+                                       generator: cf_provider
+                                   });
+                               asmsServer.publish(desiredStream, act);
+                           } else {
+                               console.log("We don't have a user name so don't raise an activity");
+                               console.dir(client.handshake.session.user);
+                           }
+
+                       } else {
+                           console.log("Got error publishing welcome message")
+                       }
+                   });
+            }
+        });
+
+
+
+		client.on('message', function(message) {
+            var actHash = {
+                actor: currentUser,
+                verb: 'post',
+                object: {objectType: "note", content: message, displayName: ""},
+                target: thisApp,
+                provider: provider,
+                generator: cf_provider
+            }
+
+            if (actHash.verb == "post") {
+                actHash.title = "posted a " + actHash.object.objectType;
+
+            }
+
+            var act = new asmsServer.Activity(actHash);
+            // Send back the message to the users room.
+            asmsServer.publish(desiredStream, act);
 		});
 
-		client.on('disconnect', function() { console.log('disconnect'); });
+		client.on('disconnect', function() {
+				console.log('********* disconnect');
+				asmsServer.unsubscribe(desiredStream);
+				console.log("unsubscribed from firehose");
+
+				if (client.handshake.session.user && client.handshake.session.user.name) {
+						asmsServer.publish(desiredStream, new asmsServer.Activity({
+								actor: currentUser,
+								verb: 'disconnect',
+								object: thisInstance,
+								target: thisApp,
+								title: "disconnected from",
+								provider: provider,
+								generator: cf_provider
+						}));
+
+
+				} else {
+						console.log("User disconnected");
+						console.dir(client.handshake);
+				}
+		});
 	});
 
 	io.sockets.on('error', function(){ console.log(arguments); });
 
 	return io;
-};
+};

package.json

@@ -6,6 +6,9 @@
   "engines" : ["node"],
   "repository" : { "type":"git", "url":"http://github.com/mape/node-express-boilerplate" },
   "dependencies" : {
+	  "activity-streams-mongoose": ">=0.0.11",
+	  "mongoose": "",
+	  "underscore": "",
     "cloudfoundry": ">=0.1.0",
     "connect" : ">=1.6.0",
     "connect-assetmanager" : ">=0.0.21",

public/css/client.css

@@ -50,18 +50,18 @@ strong {
 	height: 100%;
 	box-shadow: 0 0 5px rgba(0,0,0,0.6) inset;
 }
-#bubble {
+#stream {
 	position: absolute;
 	top: 20px;
 	left: 155px;
 	max-height: 100%;
 	overflow: hidden;
 	padding: 0 0 0 65px;
 }
-#bubble ul {
+#stream ul {
 	height: 2000px;
 }
-#bubble li {
+#stream li {
 	position:relative;
 	padding:15px;
 	margin:1em 0 3em 100px;
@@ -75,13 +75,13 @@ strong {
 	opacity: 0;
 	-vendor-transition: opacity 4s linear;
 }
-#bubble .avatar {
+#stream .avatar {
 	position: absolute;
 	right: 100%;
 	top: 10px;
 	margin-right: 45px;
 }
-#bubble .service {
+#stream .service {
 	position: absolute;
 	right: 100%;
 	top: -0px;
@@ -91,10 +91,10 @@ strong {
 	padding: 2px;
 	box-shadow: 0 0 5px #000;
 }
-#bubble li:first-child {
+#stream li:first-child {
 	opacity: 1;
 }
-#bubble li:after {
+#stream li:after {
 	content: "";
 	position: absolute;
 	bottom: -20px;

public/js/jquery.client.js

@@ -23,24 +23,43 @@
 		$$('#connected').addClass('on').find('strong').text('Online');
 	});
 
-	var image = $.trim($('#image').val());
-	var service = $.trim($('#service').val());
-	socketIoClient.on('message', function(msg) {
-		var $li = $('<li>').text(msg).append($('<img class="avatar">').attr('src', image));
-		if (service) {
-			$li.append($('<img class="service">').attr('src', service));
-		}
-		$$('#bubble ul').prepend($li);
-		$$('#bubble').scrollTop(98).stop().animate({
-			'scrollTop': '0'
-		}, 500);
-		setTimeout(function() {
-			$li.remove();
-		}, 5000);
-
-		setTimeout(function() {
-			socketIoClient.send('pong');
-		}, 1000);
+	socketIoClient.on('message', function(json) {
+    var doc = JSON.parse(json);
+    if (doc) {
+        var msg = doc.actor.displayName + " " + doc.title + " " + doc.object.displayName;
+        if (doc.target) {
+          msg+= " in " + doc.target.displayName;
+        }
+        if (doc.generator) {
+          msg+= " via " + doc.generator.displayName;
+        }
+
+        if (doc.object && doc.object.content) {
+            msg+= ": " + doc.object.content;
+        }
+
+        var $li = $('<li>').text(msg).append($('<img class="avatar">').attr('src', doc.actor.image.url));
+        if (doc.provider && doc.provider.icon && doc.provider.icon.url) {
+            $li.append($('<img class="service">').attr('src', doc.provider.icon.url));
+        }
+        $$('#stream ul').prepend($li);
+        $$('#bubble').scrollTop(98).stop().animate({
+        			'scrollTop': '0'
+        		}, 5000);
+        setTimeout(function() {
+        			$li.remove();
+        		}, 5000);
+
+        if (doc.verb == "connect") {
+            setTimeout(function() {
+                socketIoClient.send('Ok great news !');
+            }, 1000);
+        } else {
+            setTimeout(function() {
+                socketIoClient.send('I am still here');
+            }, 10000);
+        }
+    }
 	});
 
 	socketIoClient.on('disconnect', function() {

server.js

@@ -1,5 +1,7 @@
 // Fetch the site configuration
 var siteConf = require('./lib/getConfig');
+var cf = require('cloudfoundry');
+var _ = require('underscore')._;
 
 process.title = siteConf.uri.replace(/http:\/\/(www)?/, '');
 
@@ -21,12 +23,47 @@ var assetHandler = require('connect-assetmanager-handlers');
 var notifoMiddleware = require('connect-notifo');
 var DummyHelper = require('./lib/dummy-helper');
 
+var mongoose = require('mongoose');
+mongoose.connect(siteConf.mongoUrl);
+
 // Session store
 var RedisStore = require('connect-redis')(express);
 var sessionStore = new RedisStore(siteConf.redisOptions);
 
+var asmsDB = require('activity-streams-mongoose')(mongoose, {full: false, redis: siteConf.redisOptions, defaultActor: '/img/default.png'});
+
+var thisApp = new asmsDB.ActivityObject({displayName: 'Activity Streams App', url: siteConf.uri, image:{url: '/img/as-logo-sm.png'}});
+var thisInstance = {displayName: "Instance 0 -- Local"};
+if (cf.app) {
+    thisInstance.image = {url: '/img/cf-process.jpg'};
+    thisInstance.url = "http://" + cf.host + ":" + cf.port;
+    thisInstance.displayName = "App Instance " + cf.app['instance_index'] + " at " + thisInstance.url;
+    thisInstance.content = cf.app['instance_id']
+}
+
+thisApp.save(function (err) {
+    if (err === null) {
+        var startAct = new asmsDB.Activity(
+            {
+            actor: {displayName: siteConf.user_email, image:{url: "img/me.jpg"}},
+            verb: 'start',
+            object: thisInstance,
+            target: thisApp,
+            title: "started"
+            });
+
+        asmsDB.publish('firehose', startAct);
+    }
+});
+
 var app = module.exports = express.createServer();
 app.listen(siteConf.internal_port, null);
+app.asmsDB = asmsDB;
+app.siteConf = siteConf;
+app.thisApp = thisApp;
+app.thisInstance = thisInstance;
+app.cookieName = "jsessionid"; //Use this name to get sticky sessions. Default connect name is 'connect.sid';
+// Cookie name must be lowercase
 
 // Setup socket.io server
 var socketIo = new require('./lib/socket-io-server.js')(app, sessionStore);
@@ -97,7 +134,8 @@ app.configure(function() {
 	app.use(express.cookieParser());
 	app.use(assetsMiddleware);
 	app.use(express.session({
-		'store': sessionStore
+        'key': app.cookieName
+		, 'store': sessionStore
 		, 'secret': siteConf.sessionSecret
 	}));
 	app.use(express.logger({format: ':response-time ms - :date - :req[x-real-ip] - :method :url :user-agent / :referrer'}));
@@ -169,18 +207,32 @@ function NotFound(msg){
 	Error.captureStackTrace(this, arguments.callee);
 }
 
-// Routing
-app.all('/', function(req, res) {
-	// Set example session uid for use with socket.io.
+function loadUser(req, res, next) {
 	if (!req.session.uid) {
 		req.session.uid = (0 | Math.random()*1000000);
-	}
-	res.locals({
-		'key': 'value'
-	});
-	res.render('index');
+	} else if (req.session.auth){
+       if (req.session.auth.github)
+        req.providerFavicon = '//github.com/favicon.ico';
+       else if (req.session.auth.twitter)
+        req.providerFavicon = '//twitter.com/favicon.ico';
+       else if (req.session.auth.facebook)
+        req.providerFavicon = '//facebook.com/favicon.ico';
+    }
+    var displayName = req.session.user ? req.session.user.name : 'UID: '+(req.session.uid || 'has no UID');
+    var avatarUrl = ((req.session.auth && req.session.user.image) ? req.session.user.image : '/img/codercat-sm.jpg');
+    req.user = {displayName: displayName, image: {url: avatarUrl}};
+    next();
+}
+
+// Routing
+app.get('/', loadUser, function(req, res) {
+    res.render('index', {
+				currentUser: req.user,
+				providerFavicon: req.providerFavicon,
+		});
 });
 
+
 // Initiate this after all other routing is done, otherwise wildcard will go crazy.
 var dummyHelpers = new DummyHelper(app);
 
@@ -189,4 +241,4 @@ app.all('*', function(req, res){
 	throw new NotFound;
 });
 
-console.log('Running in '+(process.env.NODE_ENV || 'development')+' mode @ '+siteConf.uri);
+console.log('Running in '+(process.env.NODE_ENV || 'development')+' mode @ '+siteConf.uri);