[cleanup] realtime

frappe/__init__.py

@@ -994,3 +994,13 @@ def get_logger(module=None):
 		logging_setup_complete = True
 
 	return logging.getLogger(module or "frappe")
+
+def publish_realtime(*args, **kwargs):
+	"""Publish real-time updates
+
+	:param event: Event name, like `task_progress` etc.
+	:param message: JSON message object. For async must contain `task_id`
+	:param room: Room in which to publish update (default entire site)"""
+	import frappe.async
+
+	frappe.async.publish_realtime(*args, **kwargs)

frappe/async.py

@@ -101,7 +101,7 @@ def set_task_status(task_id, status, response=None):
 		"status": status,
 		"task_id": task_id
 	})
-	emit_via_redis("task_status_change", response, room="task:" + task_id)
+	publish_realtime("task_status_change", response, room="task:" + task_id)
 
 
 def remove_old_task_logs():
@@ -120,21 +120,45 @@ def is_file_old(file_path):
 	return ((time.time() - os.stat(file_path).st_mtime) > TASK_LOG_MAX_AGE)
 
 
+def publish_realtime(event, message, room=None, user=None, doctype=None, docname=None):
+	"""Publish real-time updates
+
+	:param event: Event name, like `task_progress` etc.
+	:param message: JSON message object. For async must contain `task_id`
+	:param room: Room in which to publish update (default entire site)
+	:param user: Transmit to user
+	:param doctype: Transmit to doctype, docname
+	:param doctype: Transmit to doctype, docname"""
+	if not room:
+		if user:
+			get_user_room(user)
+		if doctype and docname:
+			get_doc_room(doctype, docname)
+		else:
+			room = get_site_room()
+
+	emit_via_redis(event, message, room)
+
 def emit_via_redis(event, message, room=None):
+	"""Publish real-time updates via redis
+
+	:param event: Event name, like `task_progress` etc.
+	:param message: JSON message object. For async must contain `task_id`
+	:param room: name of the room"""
 	r = get_redis_server()
+
 	try:
 		r.publish('events', json.dumps({'event': event, 'message': message, 'room': room}))
 	except redis.exceptions.ConnectionError:
 		pass
 
-
 def put_log(line_no, line, task_id=None):
 	r = get_redis_server()
 	if not task_id:
 		task_id = frappe.local.task_id
 	task_progress_room = "task_progress:" + frappe.local.task_id
 	task_log_key = "task_log:" + task_id
-	emit_via_redis('task_progress', {
+	publish_realtime('task_progress', {
 		"message": {
 			"lines": {line_no: line}
 		},
@@ -195,19 +219,6 @@ def get_user_info(sid):
 		'user': session.user,
 	}
 
-def new_comment(doc, event):
-	if not doc.comment_doctype:
-		return
-	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_site_room())
-		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([frappe.local.site, ':doc:', doctype, '/', docname])
 

frappe/core/doctype/comment/comment.py

@@ -32,6 +32,21 @@ class Comment(Document):
 			"feed_type": comment_type
 		}
 
+	def after_insert(self):
+		"""Send realtime updates"""
+		if not self.comment_doctype:
+			return
+		if self.comment_doctype == 'Message':
+			if self.comment_docname == frappe.session.user:
+				message = self.as_dict()
+				message['broadcast'] = True
+				frappe.publish_realtime('new_message', message)
+			else:
+				frappe.publish_realtime('new_message', self.as_dict(), user=frappe.session.user)
+		else:
+			frappe.publish_realtime('new_comment', self.as_dict(), doctype= self.comment_doctype,
+				docname = self.comment_docname)
+
 	def validate(self):
 		"""Raise exception for more than 50 comments."""
 		if frappe.db.sql("""select count(*) from tabComment where comment_doctype=%s

frappe/desk/page/messages/messages.js

@@ -33,12 +33,29 @@ frappe.desk.pages.Messages = Class.extend({
 		this.page.wrapper.find(".layout-main-section-wrapper").addClass("col-sm-9");
 		this.page.wrapper.find(".page-title").removeClass("col-xs-6").addClass("col-xs-12");
 		this.page.wrapper.find(".page-actions").removeClass("col-xs-6").addClass("hidden-xs");
+		this.setup_realtime();
 	},
 
 	make: function() {
 		this.make_sidebar();
 	},
 
+	setup_realtime: function() {
+    	frappe.realtime.on('new_message', function(comment) {
+    		frappe.utils.notify(__("Message from {0}", [comment.comment_by_fullname]), comment.comment);
+    		if (frappe.get_route()[0] === 'messages') {
+    			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);
+    			}
+    		}
+    	});
+	},
+
 	make_sidebar: function() {
 		var me = this;
 		return frappe.call({

frappe/hooks.py

@@ -131,9 +131,6 @@ doc_events = {
 			"frappe.email.doctype.email_alert.email_alert.trigger_email_alerts"
 		],
 		"on_trash": "frappe.desk.notifications.clear_doctype_notifications"
-	},
-	"Comment": {
-		"after_insert": "frappe.async.new_comment"
 	}
 }
 

frappe/public/js/frappe/form/footer/timeline.js

@@ -294,4 +294,4 @@ frappe.ui.form.Comments = Class.extend({
 
 		return last_email;
 	}
-})
+});

frappe/public/js/frappe/model/model.js

@@ -122,6 +122,21 @@ $.extend(frappe.model, {
 		return frappe.model.docinfo[doctype] && frappe.model.docinfo[doctype][name] || null;
 	},
 
+	new_comment: function(comment) {
+		if (frappe.model.docinfo[comment.comment_doctype]
+				&& frappe.model.docinfo[comment.comment_doctype][comment.comment_docname]) {
+			var comments = frappe.model.docinfo[comment.comment_doctype][comment.comment_docname].comments;
+			var comment_exists = !!$.map(comments,
+					function(x) { return x.name == comment.name? true : undefined}).length
+			if (!comment_exists) {
+				 frappe.model.docinfo[comment.comment_doctype][comment.comment_docname].comments = comments.concat([comment]);
+			}
+		}
+		if (cur_frm.doctype === comment.comment_doctype && cur_frm.docname === comment.comment_docname) {
+				cur_frm.comments.refresh();
+		}
+	},
+
 	get_shared: function(doctype, name) {
 		return frappe.model.get_docinfo(doctype, name).shared;
 	},

frappe/public/js/frappe/socket.js

@@ -47,33 +47,7 @@ frappe.socket = {
     	frappe.socket.socket.on('task_progress', function(data) {
     	  frappe.socket.process_response(data, "progress");
     	});
-    	frappe.socket.socket.on('new_comment', function(comment) {
-    		if (frappe.model.docinfo[comment.comment_doctype] && frappe.model.docinfo[comment.comment_doctype][comment.comment_docname]) {
-    			var comments = frappe.model.docinfo[comment.comment_doctype][comment.comment_docname].comments
-    			var comment_exists = !!$.map(comments, function(x) { return x.name == comment.name? true : undefined}).length
-    			if (!comment_exists) {
-    				 frappe.model.docinfo[comment.comment_doctype][comment.comment_docname].comments = comments.concat([comment]);
-    			}
-    		}
-    		if (cur_frm.doctype === comment.comment_doctype && cur_frm.docname === comment.comment_docname) {
-    				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') {
-    			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() {
     	// subscribe again to open_tasks
@@ -112,3 +86,8 @@ frappe.socket = {
 }
 
 $(frappe.socket.init);
+
+frappe.require("frappe.realtime");
+frappe.realtime.on = function(event, callback) {
+	frappe.socket.socket.on(event, callback);
+}

frappe/public/js/frappe/ui/field_group.js

@@ -54,7 +54,7 @@ frappe.ui.FieldGroup = frappe.ui.form.Layout.extend({
 			if(f.get_parsed_value) {
 				var v = f.get_parsed_value();
 
-				if(f.df.reqd && !v)
+				if(f.df.reqd && is_null(v))
 					errors.push(__(f.df.label));
 
 				if(v) ret[f.df.fieldname] = v;

frappe/public/js/frappe/views/formview.js

@@ -19,9 +19,20 @@ frappe.views.FormFactory = frappe.views.Factory.extend({
 			me.show_doc(route);
 		}
 
-		$(document).on("page-change", function() {
-			frappe.ui.form.close_grid_form();
-		});
+
+		if(!this.initialized) {
+			$(document).on("page-change", function() {
+				frappe.ui.form.close_grid_form();
+			});
+
+			frappe.realtime.on("new_comment", function(data) {
+				frappe.model.new_comment(data);
+			});
+
+		}
+
+
+		this.initialized = true;
 	},
 	show_doc: function(route) {
 		var dt = route[1],