SocketIO push messages

frappe/async.py

@@ -9,7 +9,7 @@ import frappe
 import os
 import time
 from functools import wraps
-from frappe.utils import get_site_path
+from frappe.utils import get_site_path, get_url
 import json
 from frappe import conf
 
@@ -174,8 +174,27 @@ def can_subscribe_doc(doctype, docname, sid):
 		raise PermissionError()
 	return True
 
+@frappe.whitelist(allow_guest=True)
+def get_user_info(sid):
+	from frappe.sessions import Session
+	session = Session(None).get_session_data()
+	return {
+		'user': session.user,
+	}
+
 def new_comment(doc, event):
-	emit_via_redis('new_comment', doc.as_dict(), room=get_doc_room(doc.comment_doctype, doc.comment_docname))
+	if doc.comment_doctype == 'Message':
+		if doc.comment_docname == frappe.session.user:
+			message = doc.as_dict()
+			message['broadcast'] = True
+			emit_via_redis('new_message', message, room=get_url())
+		else:
+			emit_via_redis('new_message', doc.as_dict(), room=get_user_room(doc.comment_docname))
+	else:
+		emit_via_redis('new_comment', doc.as_dict(), room=get_doc_room(doc.comment_doctype, doc.comment_docname))
 
 def get_doc_room(doctype, docname):
 	return ''.join(['doc:', doctype, '/', docname])
+
+def get_user_room(user):
+	return ''.join(['user:', user])

frappe/desk/page/messages/messages.js

@@ -37,7 +37,6 @@ frappe.desk.pages.Messages = Class.extend({
 
 	make: function() {
 		this.make_sidebar();
-		this.set_next_refresh();
 	},
 
 	make_sidebar: function() {
@@ -155,31 +154,7 @@ frappe.desk.pages.Messages = Class.extend({
 		});
 	},
 
-	refresh: function() {
-		// check for updates every 5 seconds if page is active
-		this.set_next_refresh();
-
-		if(!frappe.session_alive) {
-			// not in session
-			return;
-		}
-
-		if(frappe.get_route()[0]!="messages") {
-			// not on messages page
-			return;
-		}
-
-		if (this.list) {
-			this.list.run();
-		}
-	},
-
-	set_next_refresh: function() {
-		// 30 seconds
-		setTimeout("frappe.desk.pages.messages.refresh()", 30000);
-	},
-
-	////
+	refresh: function() {},
 
 	get_contact: function() {
 		var route = location.hash;

frappe/public/js/frappe/desk.js

@@ -126,10 +126,6 @@ frappe.Application = Class.extend({
 
 						if(frappe.get_route()[0] != "messages") {
 							if(r.message.new_messages.length) {
-								$.each(r.message.new_messages, function(i, m) {
-									frappe.utils.notify(__("Message from {0}", [m.comment_by_fullname]),
-										m.comment);
-								});
 								frappe.utils.set_title_prefix("(" + r.message.new_messages.length + ")");
 							}
 						}

frappe/public/js/frappe/socket.js

@@ -49,6 +49,22 @@ frappe.socket = {
 			cur_frm.comments.refresh();
 		}
     });
+    frappe.socket.socket.on('new_message', function(comment) {
+		frappe.utils.notify(__("Message from {0}", [comment.comment_by_fullname]), comment.comment);
+		if ($(cur_page.page).data('page-route') === 'messages') {
+			console.log('messages page open');
+			var current_contact = $(cur_page.page).find('[data-contact]').data('contact');
+			var on_broadcast_page = current_contact === user;
+			if (current_contact == comment.owner || (on_broadcast_page && comment.broadcast)) {
+				var $row = $('<div class="list-row"/>');
+				frappe.desk.pages.messages.list.data.unshift(comment);
+				frappe.desk.pages.messages.list.render_row($row, comment);
+				frappe.desk.pages.messages.list.parent.prepend($row);
+			}
+		}
+		else {
+		}
+    });
 
   },
   setup_reconnect: function() {

socketio.js

@@ -14,6 +14,22 @@ app.get('/', function(req, res){
 });
 
 io.on('connection', function(socket){
+	socket.join(socket.request.headers.origin);
+	var sid = cookie.parse(socket.request.headers.cookie).sid
+	if(!sid) {
+		return;
+	}
+	request.post('http://localhost:8000/api/method/frappe.async.get_user_info')
+		.type('form')
+		.send({
+			sid: sid
+		})
+		.end(function(err, res) {
+			if(res.status == 200) {
+				var room = get_user_room(res.body.message.user);
+				socket.join(room);
+			}
+		})
 	socket.on('task_subscribe', function(task_id) {
 		var room = 'task:' + task_id;
 		socket.join(room);
@@ -24,10 +40,6 @@ io.on('connection', function(socket){
 		send_existing_lines(task_id, socket);
 	})
 	socket.on('doc_subscribe', function(doctype, docname) {
-		var sid = cookie.parse(socket.request.headers.cookie).sid
-		if(!sid) {
-			return;
-		}
 		request.post('http://localhost:8000/api/method/frappe.async.can_subscribe_doc')
 			.type('form')
 			.send({
@@ -63,6 +75,7 @@ function send_existing_lines(task_id, socket) {
 subscriber.on("message", function(channel, message) {
 	message = JSON.parse(message);
 	io.to(message.room).emit(message.event, message.message);
+	console.log(message.room, message.event, message.message)
 });
 
 subscriber.subscribe("events");
@@ -74,3 +87,7 @@ http.listen(3000, function(){
 function get_doc_room(doctype, docname) {
 	return 'doc:'+ doctype + '/' + docname;
 }
+
+function get_user_room(user) {
+	return 'user:' + user;
+}