Process message events more similarly to how other events are handled.

(imported from commit 5ebcd7a10fe54a8ecafb9550348a1e0418d4cbda)

zerver/lib/actions.py

@@ -50,9 +50,7 @@ from zerver.lib.push_notifications import num_push_devices_for_user, \
      send_apple_push_notification, send_android_push_notification
 from zerver.lib.notifications import clear_followup_emails_queue
 from zerver.lib.narrow import check_supported_events_narrow_filter
-
 from zerver.tornado_callbacks import send_event
-from zerver import tornado_callbacks
 
 import DNS
 import ujson
@@ -364,16 +362,16 @@ def do_send_messages(messages):
             if user_profile.email in user_presences:
                 presences[user_profile.id] = user_presences[user_profile.email]
 
-        data = dict(
-            type         = 'new_message',
+        event = dict(
+            type         = 'message',
             message      = message['message'].id,
             message_dict_markdown = message['message'].to_dict(apply_markdown=True),
             message_dict_no_markdown = message['message'].to_dict(apply_markdown=False),
-            presences    = presences,
+            presences    = presences)
         users = [{'id': user.id,
                   'flags': user_flags.get(user.id, []),
                   'always_push_notify': always_push_notify(user)}
-                            for user in message['active_recipients']])
+                 for user in message['active_recipients']]
         if message['message'].recipient.type == Recipient.STREAM:
             # Note: This is where authorization for single-stream
             # get_updates happens! We only attach stream data to the
@@ -383,15 +381,15 @@ def do_send_messages(messages):
             if message['stream'] is None:
                 message['stream'] = Stream.objects.select_related("realm").get(id=message['message'].recipient.type_id)
             if message['stream'].is_public():
-                data['realm_id'] = message['stream'].realm.id
-                data['stream_name'] = message['stream'].name
+                event['realm_id'] = message['stream'].realm.id
+                event['stream_name'] = message['stream'].name
             if message['stream'].invite_only:
-                data['invite_only'] = True
+                event['invite_only'] = True
         if message['local_id'] is not None:
-            data['local_id'] = message['local_id']
+            event['local_id'] = message['local_id']
         if message['sender_queue_id'] is not None:
-            data['sender_queue_id'] = message['sender_queue_id']
-        tornado_callbacks.send_notification(data)
+            event['sender_queue_id'] = message['sender_queue_id']
+        send_event(event, users)
         if (settings.ENABLE_FEEDBACK and
             message['message'].recipient.type == Recipient.PERSONAL and
             settings.FEEDBACK_BOT in [up.email for up in message['recipients']]):

zerver/tests.py

@@ -2293,7 +2293,7 @@ class SubscriptionAPITest(AuthedTestCase):
         self.assert_length(queries, 37)
 
         self.assert_length(events, 4, exact=True)
-        for ev in filter(lambda x: 'message' not in x, events):
+        for ev in filter(lambda x: x['event']['type'] != 'message', events):
             self.assertEqual(ev['event']['op'], 'add')
             self.assertEqual(
                     set(ev['event']['subscriptions'][0]['subscribers']),

zerver/tornado_callbacks.py

@@ -93,7 +93,7 @@ def receiver_is_idle(user_profile_id, realm_presences):
 
     return off_zulip or idle_too_long
 
-def process_new_message(event_template):
+def process_message_event(event_template, users):
     realm_presences = event_template['presences']
     sender_queue_id = event_template.get('sender_queue_id', None)
     if "message_dict_markdown" in event_template:
@@ -122,7 +122,7 @@ def process_new_message(event_template):
             if sender_queue_id is not None and client.event_queue.id == sender_queue_id:
                 send_to_clients[client.event_queue.id]['is_sender'] = True
 
-    for user_data in event_template['users']:
+    for user_data in users:
         user_profile_id = user_data['id']
         flags = user_data.get('flags', [])
 
@@ -208,13 +208,13 @@ def process_userdata_event(event_template, users):
             if client.accepts_event(user_event):
                 client.add_event(user_event)
 
-def process_notification(data):
-    event = data['event']
-    users = data['users']
+def process_notification(notice):
+    event = notice['event']
+    users = notice['users']
     if event['type'] in ["update_message"]:
         process_userdata_event(event, users)
     elif event['type'] == "message":
-        process_new_message(data)
+        process_message_event(event, users)
     else:
         process_event(event, users)
 
@@ -235,4 +235,6 @@ def send_notification(data):
     return queue_json_publish("notify_tornado", data, send_notification_http)
 
 def send_event(event, users):
-    return send_notification(dict(event=event, users=users))
+    return queue_json_publish("notify_tornado",
+                              dict(event=event, users=users),
+                              send_notification_http)