mirror of
https://github.com/zulip/zulip.git
synced 2025-11-05 14:35:27 +00:00
c7e3c3ce38
Previously, client descriptors were referenced directly from the handler object. Once we split the Tornado process into separate queue and connection servers, these will no longer be in the same process, so we need to reference them by ID instead.
759 lines
31 KiB
Python
759 lines
31 KiB
Python
from __future__ import absolute_import
|
|
|
|
from django.conf import settings
|
|
from django.utils.timezone import now
|
|
from collections import deque
|
|
import datetime
|
|
import os
|
|
import time
|
|
import socket
|
|
import logging
|
|
import ujson
|
|
import requests
|
|
import six.moves.cPickle as pickle
|
|
import atexit
|
|
import sys
|
|
import signal
|
|
import tornado
|
|
import random
|
|
import traceback
|
|
from zerver.decorator import RespondAsynchronously, JsonableError
|
|
from zerver.lib.cache import cache_get_many, message_cache_key, \
|
|
user_profile_by_id_cache_key, cache_save_user_profile
|
|
from zerver.lib.cache_helpers import cache_with_key
|
|
from zerver.lib.handlers import get_handler_by_id, finish_handler
|
|
from zerver.lib.utils import statsd
|
|
from zerver.middleware import async_request_restart
|
|
from zerver.models import get_client, Message
|
|
from zerver.lib.narrow import build_narrow_filter
|
|
from zerver.lib.queue import queue_json_publish
|
|
from zerver.lib.response import json_success, json_error
|
|
from zerver.lib.timestamp import timestamp_to_datetime
|
|
import copy
|
|
|
|
# The idle timeout used to be a week, but we found that in that
|
|
# situation, queues from dead browser sessions would grow quite large
|
|
# due to the accumulation of message data in those queues.
|
|
IDLE_EVENT_QUEUE_TIMEOUT_SECS = 60 * 10
|
|
EVENT_QUEUE_GC_FREQ_MSECS = 1000 * 60 * 5
|
|
|
|
# Capped limit for how long a client can request an event queue
|
|
# to live
|
|
MAX_QUEUE_TIMEOUT_SECS = 7 * 24 * 60 * 60
|
|
|
|
# The heartbeats effectively act as a server-side timeout for
|
|
# get_events(). The actual timeout value is randomized for each
|
|
# client connection based on the below value. We ensure that the
|
|
# maximum timeout value is 55 seconds, to deal with crappy home
|
|
# wireless routers that kill "inactive" http connections.
|
|
HEARTBEAT_MIN_FREQ_SECS = 45
|
|
|
|
class ClientDescriptor(object):
|
|
def __init__(self, user_profile_id, user_profile_email, realm_id, event_queue,
|
|
event_types, client_type, apply_markdown=True,
|
|
all_public_streams=False, lifespan_secs=0, narrow=[]):
|
|
# These objects are serialized on shutdown and restored on restart.
|
|
# If fields are added or semantics are changed, temporary code must be
|
|
# added to load_event_queues() to update the restored objects.
|
|
# Additionally, the to_dict and from_dict methods must be updated
|
|
self.user_profile_id = user_profile_id
|
|
self.user_profile_email = user_profile_email
|
|
self.realm_id = realm_id
|
|
self.current_handler_id = None
|
|
self.event_queue = event_queue
|
|
self.queue_timeout = lifespan_secs
|
|
self.event_types = event_types
|
|
self.last_connection_time = time.time()
|
|
self.apply_markdown = apply_markdown
|
|
self.all_public_streams = all_public_streams
|
|
self.client_type = client_type
|
|
self._timeout_handle = None
|
|
self.narrow = narrow
|
|
self.narrow_filter = build_narrow_filter(narrow)
|
|
|
|
# Clamp queue_timeout to between minimum and maximum timeouts
|
|
self.queue_timeout = max(IDLE_EVENT_QUEUE_TIMEOUT_SECS, min(self.queue_timeout, MAX_QUEUE_TIMEOUT_SECS))
|
|
|
|
def to_dict(self):
|
|
# If you add a new key to this dict, make sure you add appropriate
|
|
# migration code in from_dict or load_event_queues to account for
|
|
# loading event queues that lack that key.
|
|
return dict(user_profile_id=self.user_profile_id,
|
|
user_profile_email=self.user_profile_email,
|
|
realm_id=self.realm_id,
|
|
event_queue=self.event_queue.to_dict(),
|
|
queue_timeout=self.queue_timeout,
|
|
event_types=self.event_types,
|
|
last_connection_time=self.last_connection_time,
|
|
apply_markdown=self.apply_markdown,
|
|
all_public_streams=self.all_public_streams,
|
|
narrow=self.narrow,
|
|
client_type=self.client_type.name)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d):
|
|
if 'user_profile_email' not in d:
|
|
# Temporary migration for the addition of the new user_profile_email field
|
|
from zerver.models import get_user_profile_by_id
|
|
d['user_profile_email'] = get_user_profile_by_id(d['user_profile_id']).email
|
|
ret = cls(d['user_profile_id'], d['user_profile_email'], d['realm_id'],
|
|
EventQueue.from_dict(d['event_queue']), d['event_types'],
|
|
get_client(d['client_type']), d['apply_markdown'], d['all_public_streams'],
|
|
d['queue_timeout'], d.get('narrow', []))
|
|
ret.last_connection_time = d['last_connection_time']
|
|
return ret
|
|
|
|
def prepare_for_pickling(self):
|
|
self.current_handler_id = None
|
|
self._timeout_handle = None
|
|
|
|
def add_event(self, event):
|
|
if self.current_handler_id is not None:
|
|
handler = get_handler_by_id(self.current_handler_id)
|
|
async_request_restart(handler._request)
|
|
|
|
self.event_queue.push(event)
|
|
self.finish_current_handler()
|
|
|
|
def finish_current_handler(self):
|
|
if self.current_handler_id is not None:
|
|
err_msg = "Got error finishing handler for queue %s" % (self.event_queue.id,)
|
|
try:
|
|
finish_handler(self.current_handler_id, self.event_queue.id,
|
|
self.event_queue.contents(), self.apply_markdown)
|
|
except Exception:
|
|
logging.exception(err_msg)
|
|
finally:
|
|
self.disconnect_handler()
|
|
return True
|
|
return False
|
|
|
|
def accepts_event(self, event):
|
|
if self.event_types is not None and event["type"] not in self.event_types:
|
|
return False
|
|
if event["type"] == "message":
|
|
return self.narrow_filter(event)
|
|
return True
|
|
|
|
# TODO: Refactor so we don't need this function
|
|
def accepts_messages(self):
|
|
return self.event_types is None or "message" in self.event_types
|
|
|
|
def idle(self, now):
|
|
if not hasattr(self, 'queue_timeout'):
|
|
self.queue_timeout = IDLE_EVENT_QUEUE_TIMEOUT_SECS
|
|
|
|
return (self.current_handler_id is None
|
|
and now - self.last_connection_time >= self.queue_timeout)
|
|
|
|
def connect_handler(self, handler_id):
|
|
self.current_handler_id = handler_id
|
|
set_descriptor_by_handler_id(handler_id, self)
|
|
self.last_connection_time = time.time()
|
|
def timeout_callback():
|
|
self._timeout_handle = None
|
|
# All clients get heartbeat events
|
|
self.add_event(dict(type='heartbeat'))
|
|
ioloop = tornado.ioloop.IOLoop.instance()
|
|
heartbeat_time = time.time() + HEARTBEAT_MIN_FREQ_SECS + random.randint(0, 10)
|
|
if self.client_type.name != 'API: heartbeat test':
|
|
self._timeout_handle = ioloop.add_timeout(heartbeat_time, timeout_callback)
|
|
|
|
def disconnect_handler(self, client_closed=False):
|
|
if self.current_handler_id:
|
|
handler = get_handler_by_id(self.current_handler_id)
|
|
request = handler._request
|
|
delete_descriptor_by_handler_id(self.current_handler_id, None)
|
|
if client_closed:
|
|
logging.info("Client disconnected for queue %s (%s via %s)" %
|
|
(self.event_queue.id, self.user_profile_email,
|
|
request.client.name))
|
|
self.current_handler_id = None
|
|
if self._timeout_handle is not None:
|
|
ioloop = tornado.ioloop.IOLoop.instance()
|
|
ioloop.remove_timeout(self._timeout_handle)
|
|
self._timeout_handle = None
|
|
|
|
def cleanup(self):
|
|
do_gc_event_queues([self.event_queue.id], [self.user_profile_id],
|
|
[self.realm_id])
|
|
|
|
descriptors_by_handler_id = {}
|
|
|
|
def get_descriptor_by_handler_id(handler_id):
|
|
return descriptors_by_handler_id.get(handler_id)
|
|
|
|
def set_descriptor_by_handler_id(handler_id, client_descriptor):
|
|
descriptors_by_handler_id[handler_id] = client_descriptor
|
|
|
|
def delete_descriptor_by_handler_id(handler_id, client_descriptor):
|
|
del descriptors_by_handler_id[handler_id]
|
|
|
|
def compute_full_event_type(event):
|
|
if event["type"] == "update_message_flags":
|
|
if event["all"]:
|
|
# Put the "all" case in its own category
|
|
return "all_flags/%s/%s" % (event["flag"], event["operation"])
|
|
return "flags/%s/%s" % (event["operation"], event["flag"])
|
|
return event["type"]
|
|
|
|
class EventQueue(object):
|
|
def __init__(self, id):
|
|
self.queue = deque()
|
|
self.next_event_id = 0
|
|
self.id = id
|
|
self.virtual_events = {}
|
|
|
|
def to_dict(self):
|
|
# If you add a new key to this dict, make sure you add appropriate
|
|
# migration code in from_dict or load_event_queues to account for
|
|
# loading event queues that lack that key.
|
|
return dict(id=self.id,
|
|
next_event_id=self.next_event_id,
|
|
queue=list(self.queue),
|
|
virtual_events=self.virtual_events)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d):
|
|
ret = cls(d['id'])
|
|
ret.next_event_id = d['next_event_id']
|
|
ret.queue = deque(d['queue'])
|
|
ret.virtual_events = d.get("virtual_events", {})
|
|
return ret
|
|
|
|
def push(self, event):
|
|
event['id'] = self.next_event_id
|
|
self.next_event_id += 1
|
|
full_event_type = compute_full_event_type(event)
|
|
if (full_event_type in ["pointer", "restart"] or
|
|
full_event_type.startswith("flags/")):
|
|
if full_event_type not in self.virtual_events:
|
|
self.virtual_events[full_event_type] = copy.deepcopy(event)
|
|
return
|
|
# Update the virtual event with the values from the event
|
|
virtual_event = self.virtual_events[full_event_type]
|
|
virtual_event["id"] = event["id"]
|
|
if "timestamp" in event:
|
|
virtual_event["timestamp"] = event["timestamp"]
|
|
if full_event_type == "pointer":
|
|
virtual_event["pointer"] = event["pointer"]
|
|
elif full_event_type == "restart":
|
|
virtual_event["server_generation"] = event["server_generation"]
|
|
elif full_event_type.startswith("flags/"):
|
|
virtual_event["messages"] += event["messages"]
|
|
else:
|
|
self.queue.append(event)
|
|
|
|
# Note that pop ignores virtual events. This is fine in our
|
|
# current usage since virtual events should always be resolved to
|
|
# a real event before being given to users.
|
|
def pop(self):
|
|
return self.queue.popleft()
|
|
|
|
def empty(self):
|
|
return len(self.queue) == 0 and len(self.virtual_events) == 0
|
|
|
|
# See the comment on pop; that applies here as well
|
|
def prune(self, through_id):
|
|
while len(self.queue) != 0 and self.queue[0]['id'] <= through_id:
|
|
self.pop()
|
|
|
|
def contents(self):
|
|
contents = []
|
|
virtual_id_map = {}
|
|
for event_type in self.virtual_events:
|
|
virtual_id_map[self.virtual_events[event_type]["id"]] = self.virtual_events[event_type]
|
|
virtual_ids = sorted(list(virtual_id_map.keys()))
|
|
|
|
# Merge the virtual events into their final place in the queue
|
|
index = 0
|
|
length = len(virtual_ids)
|
|
for event in self.queue:
|
|
while index < length and virtual_ids[index] < event["id"]:
|
|
contents.append(virtual_id_map[virtual_ids[index]])
|
|
index += 1
|
|
contents.append(event)
|
|
while index < length:
|
|
contents.append(virtual_id_map[virtual_ids[index]])
|
|
index += 1
|
|
|
|
self.virtual_events = {}
|
|
self.queue = deque(contents)
|
|
return contents
|
|
|
|
# maps queue ids to client descriptors
|
|
clients = {}
|
|
# maps user id to list of client descriptors
|
|
user_clients = {}
|
|
# maps realm id to list of client descriptors with all_public_streams=True
|
|
realm_clients_all_streams = {}
|
|
|
|
# list of registered gc hooks.
|
|
# each one will be called with a user profile id, queue, and bool
|
|
# last_for_client that is true if this is the last queue pertaining
|
|
# to this user_profile_id
|
|
# that is about to be deleted
|
|
gc_hooks = []
|
|
|
|
next_queue_id = 0
|
|
|
|
def add_client_gc_hook(hook):
|
|
gc_hooks.append(hook)
|
|
|
|
def get_client_descriptor(queue_id):
|
|
return clients.get(queue_id)
|
|
|
|
def get_client_descriptors_for_user(user_profile_id):
|
|
return user_clients.get(user_profile_id, [])
|
|
|
|
def get_client_descriptors_for_realm_all_streams(realm_id):
|
|
return realm_clients_all_streams.get(realm_id, [])
|
|
|
|
def add_to_client_dicts(client):
|
|
user_clients.setdefault(client.user_profile_id, []).append(client)
|
|
if client.all_public_streams or client.narrow != []:
|
|
realm_clients_all_streams.setdefault(client.realm_id, []).append(client)
|
|
|
|
def allocate_client_descriptor(user_profile_id, user_profile_email, realm_id,
|
|
event_types, client_type, apply_markdown,
|
|
all_public_streams, lifespan_secs, narrow=[]):
|
|
global next_queue_id
|
|
id = str(settings.SERVER_GENERATION) + ':' + str(next_queue_id)
|
|
next_queue_id += 1
|
|
client = ClientDescriptor(user_profile_id, user_profile_email, realm_id,
|
|
EventQueue(id), event_types, client_type,
|
|
apply_markdown, all_public_streams, lifespan_secs, narrow)
|
|
clients[id] = client
|
|
add_to_client_dicts(client)
|
|
return client
|
|
|
|
def do_gc_event_queues(to_remove, affected_users, affected_realms):
|
|
def filter_client_dict(client_dict, key):
|
|
if key not in client_dict:
|
|
return
|
|
|
|
new_client_list = [c for c in client_dict[key] if c.event_queue.id not in to_remove]
|
|
if len(new_client_list) == 0:
|
|
del client_dict[key]
|
|
else:
|
|
client_dict[key] = new_client_list
|
|
|
|
for user_id in affected_users:
|
|
filter_client_dict(user_clients, user_id)
|
|
|
|
for realm_id in affected_realms:
|
|
filter_client_dict(realm_clients_all_streams, realm_id)
|
|
|
|
for id in to_remove:
|
|
for cb in gc_hooks:
|
|
cb(clients[id].user_profile_id, clients[id], clients[id].user_profile_id not in user_clients)
|
|
del clients[id]
|
|
|
|
def gc_event_queues():
|
|
start = time.time()
|
|
to_remove = set()
|
|
affected_users = set()
|
|
affected_realms = set()
|
|
for (id, client) in clients.iteritems():
|
|
if client.idle(start):
|
|
to_remove.add(id)
|
|
affected_users.add(client.user_profile_id)
|
|
affected_realms.add(client.realm_id)
|
|
|
|
do_gc_event_queues(to_remove, affected_users, affected_realms)
|
|
|
|
logging.info(('Tornado removed %d idle event queues owned by %d users in %.3fs.'
|
|
+ ' Now %d active queues')
|
|
% (len(to_remove), len(affected_users), time.time() - start,
|
|
len(clients)))
|
|
statsd.gauge('tornado.active_queues', len(clients))
|
|
statsd.gauge('tornado.active_users', len(user_clients))
|
|
|
|
def dump_event_queues():
|
|
start = time.time()
|
|
|
|
with file(settings.JSON_PERSISTENT_QUEUE_FILENAME, "w") as stored_queues:
|
|
ujson.dump([(qid, client.to_dict()) for (qid, client) in clients.iteritems()],
|
|
stored_queues)
|
|
|
|
logging.info('Tornado dumped %d event queues in %.3fs'
|
|
% (len(clients), time.time() - start))
|
|
|
|
def load_event_queues():
|
|
global clients
|
|
start = time.time()
|
|
|
|
if os.path.exists(settings.PERSISTENT_QUEUE_FILENAME):
|
|
try:
|
|
with file(settings.PERSISTENT_QUEUE_FILENAME, "r") as stored_queues:
|
|
clients = pickle.load(stored_queues)
|
|
except (IOError, EOFError):
|
|
pass
|
|
else:
|
|
# ujson chokes on bad input pretty easily. We separate out the actual
|
|
# file reading from the loading so that we don't silently fail if we get
|
|
# bad input.
|
|
try:
|
|
with file(settings.JSON_PERSISTENT_QUEUE_FILENAME, "r") as stored_queues:
|
|
json_data = stored_queues.read()
|
|
try:
|
|
clients = dict((qid, ClientDescriptor.from_dict(client))
|
|
for (qid, client) in ujson.loads(json_data))
|
|
except Exception:
|
|
logging.exception("Could not deserialize event queues")
|
|
except (IOError, EOFError):
|
|
pass
|
|
|
|
for client in clients.itervalues():
|
|
# Put code for migrations due to event queue data format changes here
|
|
|
|
add_to_client_dicts(client)
|
|
|
|
logging.info('Tornado loaded %d event queues in %.3fs'
|
|
% (len(clients), time.time() - start))
|
|
|
|
def send_restart_events():
|
|
event = dict(type='restart', server_generation=settings.SERVER_GENERATION)
|
|
for client in clients.itervalues():
|
|
if client.accepts_event(event):
|
|
client.add_event(event.copy())
|
|
|
|
def setup_event_queue():
|
|
load_event_queues()
|
|
atexit.register(dump_event_queues)
|
|
# Make sure we dump event queues even if we exit via signal
|
|
signal.signal(signal.SIGTERM, lambda signum, stack: sys.exit(1))
|
|
|
|
try:
|
|
os.rename(settings.PERSISTENT_QUEUE_FILENAME, "/var/tmp/event_queues.pickle.last")
|
|
except OSError:
|
|
pass
|
|
|
|
try:
|
|
os.rename(settings.JSON_PERSISTENT_QUEUE_FILENAME, "/var/tmp/event_queues.json.last")
|
|
except OSError:
|
|
pass
|
|
|
|
# Set up event queue garbage collection
|
|
ioloop = tornado.ioloop.IOLoop.instance()
|
|
pc = tornado.ioloop.PeriodicCallback(gc_event_queues,
|
|
EVENT_QUEUE_GC_FREQ_MSECS, ioloop)
|
|
pc.start()
|
|
|
|
send_restart_events()
|
|
|
|
def fetch_events(user_profile_id, user_profile_realm_id, user_profile_email,
|
|
queue_id, last_event_id, event_types, user_client, apply_markdown,
|
|
all_public_streams, lifespan_secs, narrow, dont_block, handler_id):
|
|
was_connected = False
|
|
orig_queue_id = queue_id
|
|
extra_log_data = ""
|
|
if queue_id is None:
|
|
if dont_block:
|
|
client = allocate_client_descriptor(user_profile_id, user_profile_email,
|
|
user_profile_realm_id,
|
|
event_types, user_client, apply_markdown,
|
|
all_public_streams, lifespan_secs,
|
|
narrow=narrow)
|
|
queue_id = client.event_queue.id
|
|
else:
|
|
raise JsonableError("Missing 'queue_id' argument")
|
|
else:
|
|
if last_event_id is None:
|
|
raise JsonableError("Missing 'last_event_id' argument")
|
|
client = get_client_descriptor(queue_id)
|
|
if client is None:
|
|
raise JsonableError("Bad event queue id: %s" % (queue_id,))
|
|
if user_profile_id != client.user_profile_id:
|
|
raise JsonableError("You are not authorized to get events from this queue")
|
|
client.event_queue.prune(last_event_id)
|
|
was_connected = client.finish_current_handler()
|
|
|
|
if not client.event_queue.empty() or dont_block:
|
|
ret = {'events': client.event_queue.contents()}
|
|
if orig_queue_id is None:
|
|
ret['queue_id'] = queue_id
|
|
extra_log_data = "[%s/%s]" % (queue_id, len(ret["events"]))
|
|
if was_connected:
|
|
extra_log_data += " [was connected]"
|
|
return (json_success(ret), extra_log_data)
|
|
|
|
if was_connected:
|
|
logging.info("Disconnected handler for queue %s (%s/%s)" % (queue_id, user_profile_email,
|
|
user_client.name))
|
|
client.connect_handler(handler_id)
|
|
return (RespondAsynchronously, None)
|
|
|
|
# The following functions are called from Django
|
|
|
|
# Workaround to support the Python-requests 1.0 transition of .json
|
|
# from a property to a function
|
|
requests_json_is_function = callable(requests.Response.json)
|
|
def extract_json_response(resp):
|
|
if requests_json_is_function:
|
|
return resp.json()
|
|
else:
|
|
return resp.json
|
|
|
|
def request_event_queue(user_profile, user_client, apply_markdown,
|
|
queue_lifespan_secs, event_types=None, all_public_streams=False,
|
|
narrow=[]):
|
|
if settings.TORNADO_SERVER:
|
|
req = {'dont_block' : 'true',
|
|
'apply_markdown': ujson.dumps(apply_markdown),
|
|
'all_public_streams': ujson.dumps(all_public_streams),
|
|
'client' : 'internal',
|
|
'user_client' : user_client.name,
|
|
'narrow' : ujson.dumps(narrow),
|
|
'lifespan_secs' : queue_lifespan_secs}
|
|
if event_types is not None:
|
|
req['event_types'] = ujson.dumps(event_types)
|
|
resp = requests.get(settings.TORNADO_SERVER + '/api/v1/events',
|
|
auth=requests.auth.HTTPBasicAuth(user_profile.email,
|
|
user_profile.api_key),
|
|
params=req)
|
|
|
|
resp.raise_for_status()
|
|
|
|
return extract_json_response(resp)['queue_id']
|
|
|
|
return None
|
|
|
|
def get_user_events(user_profile, queue_id, last_event_id):
|
|
if settings.TORNADO_SERVER:
|
|
resp = requests.get(settings.TORNADO_SERVER + '/api/v1/events',
|
|
auth=requests.auth.HTTPBasicAuth(user_profile.email,
|
|
user_profile.api_key),
|
|
params={'queue_id' : queue_id,
|
|
'last_event_id': last_event_id,
|
|
'dont_block' : 'true',
|
|
'client' : 'internal'})
|
|
|
|
resp.raise_for_status()
|
|
|
|
return extract_json_response(resp)['events']
|
|
|
|
|
|
# Send email notifications to idle users
|
|
# after they are idle for 1 hour
|
|
NOTIFY_AFTER_IDLE_HOURS = 1
|
|
|
|
def build_offline_notification(user_profile_id, message_id):
|
|
return {"user_profile_id": user_profile_id,
|
|
"message_id": message_id,
|
|
"timestamp": time.time()}
|
|
|
|
def missedmessage_hook(user_profile_id, queue, last_for_client):
|
|
# Only process missedmessage hook when the last queue for a
|
|
# client has been garbage collected
|
|
if not last_for_client:
|
|
return
|
|
|
|
message_ids_to_notify = []
|
|
for event in queue.event_queue.contents():
|
|
if not event['type'] == 'message' or not event['flags']:
|
|
continue
|
|
|
|
if 'mentioned' in event['flags'] and not 'read' in event['flags']:
|
|
notify_info = dict(message_id=event['message']['id'])
|
|
|
|
if not event.get('push_notified', False):
|
|
notify_info['send_push'] = True
|
|
if not event.get('email_notified', False):
|
|
notify_info['send_email'] = True
|
|
message_ids_to_notify.append(notify_info)
|
|
|
|
for notify_info in message_ids_to_notify:
|
|
msg_id = notify_info['message_id']
|
|
notice = build_offline_notification(user_profile_id, msg_id)
|
|
if notify_info.get('send_push', False):
|
|
queue_json_publish("missedmessage_mobile_notifications", notice, lambda notice: None)
|
|
if notify_info.get('send_email', False):
|
|
queue_json_publish("missedmessage_emails", notice, lambda notice: None)
|
|
|
|
@cache_with_key(message_cache_key, timeout=3600*24)
|
|
def get_message_by_id_dbwarn(message_id):
|
|
if not settings.TEST_SUITE:
|
|
logging.warning("Tornado failed to load message from memcached when delivering!")
|
|
return Message.objects.select_related().get(id=message_id)
|
|
|
|
def receiver_is_idle(user_profile_id, realm_presences):
|
|
# If a user has no message-receiving event queues, they've got no open zulip
|
|
# session so we notify them
|
|
all_client_descriptors = get_client_descriptors_for_user(user_profile_id)
|
|
message_event_queues = [client for client in all_client_descriptors if client.accepts_messages()]
|
|
off_zulip = len(message_event_queues) == 0
|
|
|
|
# It's possible a recipient is not in the realm of a sender. We don't have
|
|
# presence information in this case (and it's hard to get without an additional
|
|
# db query) so we simply don't try to guess if this cross-realm recipient
|
|
# has been idle for too long
|
|
if realm_presences is None or not user_profile_id in realm_presences:
|
|
return off_zulip
|
|
|
|
# We want to find the newest "active" presence entity and compare that to the
|
|
# activity expiry threshold.
|
|
user_presence = realm_presences[user_profile_id]
|
|
latest_active_timestamp = None
|
|
idle = False
|
|
|
|
for client, status in user_presence.iteritems():
|
|
if (latest_active_timestamp is None or status['timestamp'] > latest_active_timestamp) and \
|
|
status['status'] == 'active':
|
|
latest_active_timestamp = status['timestamp']
|
|
|
|
if latest_active_timestamp is None:
|
|
idle = True
|
|
else:
|
|
active_datetime = timestamp_to_datetime(latest_active_timestamp)
|
|
# 140 seconds is consistent with activity.js:OFFLINE_THRESHOLD_SECS
|
|
idle = now() - active_datetime > datetime.timedelta(seconds=140)
|
|
|
|
return off_zulip or idle
|
|
|
|
def process_message_event(event_template, users):
|
|
realm_presences = {int(k): v for k, v in event_template['presences'].items()}
|
|
sender_queue_id = event_template.get('sender_queue_id', None)
|
|
if "message_dict_markdown" in event_template:
|
|
message_dict_markdown = event_template['message_dict_markdown']
|
|
message_dict_no_markdown = event_template['message_dict_no_markdown']
|
|
else:
|
|
# We can delete this and get_message_by_id_dbwarn after the
|
|
# next prod deploy
|
|
message = get_message_by_id_dbwarn(event_template['message'])
|
|
message_dict_markdown = message.to_dict(True)
|
|
message_dict_no_markdown = message.to_dict(False)
|
|
sender_id = message_dict_markdown['sender_id']
|
|
message_id = message_dict_markdown['id']
|
|
message_type = message_dict_markdown['type']
|
|
sending_client = message_dict_markdown['client']
|
|
|
|
# To remove duplicate clients: Maps queue ID to {'client': Client, 'flags': flags}
|
|
send_to_clients = dict()
|
|
|
|
# Extra user-specific data to include
|
|
extra_user_data = {}
|
|
|
|
if 'stream_name' in event_template and not event_template.get("invite_only"):
|
|
for client in get_client_descriptors_for_realm_all_streams(event_template['realm_id']):
|
|
send_to_clients[client.event_queue.id] = {'client': client, 'flags': None}
|
|
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 users:
|
|
user_profile_id = user_data['id']
|
|
flags = user_data.get('flags', [])
|
|
|
|
for client in get_client_descriptors_for_user(user_profile_id):
|
|
send_to_clients[client.event_queue.id] = {'client': client, 'flags': flags}
|
|
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
|
|
|
|
# If the recipient was offline and the message was a single or group PM to him
|
|
# or she was @-notified potentially notify more immediately
|
|
received_pm = message_type == "private" and user_profile_id != sender_id
|
|
mentioned = 'mentioned' in flags
|
|
idle = receiver_is_idle(user_profile_id, realm_presences)
|
|
always_push_notify = user_data.get('always_push_notify', False)
|
|
if (received_pm or mentioned) and (idle or always_push_notify):
|
|
notice = build_offline_notification(user_profile_id, message_id)
|
|
queue_json_publish("missedmessage_mobile_notifications", notice, lambda notice: None)
|
|
notified = dict(push_notified=True)
|
|
# Don't send missed message emails if always_push_notify is True
|
|
if idle:
|
|
# We require RabbitMQ to do this, as we can't call the email handler
|
|
# from the Tornado process. So if there's no rabbitmq support do nothing
|
|
queue_json_publish("missedmessage_emails", notice, lambda notice: None)
|
|
notified['email_notified'] = True
|
|
|
|
extra_user_data[user_profile_id] = notified
|
|
|
|
for client_data in send_to_clients.itervalues():
|
|
client = client_data['client']
|
|
flags = client_data['flags']
|
|
is_sender = client_data.get('is_sender', False)
|
|
extra_data = extra_user_data.get(client.user_profile_id, None)
|
|
|
|
if not client.accepts_messages():
|
|
# The actual check is the accepts_event() check below;
|
|
# this line is just an optimization to avoid copying
|
|
# message data unnecessarily
|
|
continue
|
|
|
|
if client.apply_markdown:
|
|
message_dict = message_dict_markdown
|
|
else:
|
|
message_dict = message_dict_no_markdown
|
|
|
|
# Make sure Zephyr mirroring bots know whether stream is invite-only
|
|
if "mirror" in client.client_type.name and event_template.get("invite_only"):
|
|
message_dict = message_dict.copy()
|
|
message_dict["invite_only_stream"] = True
|
|
|
|
user_event = dict(type='message', message=message_dict, flags=flags)
|
|
if extra_data is not None:
|
|
user_event.update(extra_data)
|
|
|
|
if is_sender:
|
|
local_message_id = event_template.get('local_id', None)
|
|
if local_message_id is not None:
|
|
user_event["local_message_id"] = local_message_id
|
|
|
|
if not client.accepts_event(user_event):
|
|
continue
|
|
|
|
# The below prevents (Zephyr) mirroring loops.
|
|
if ('mirror' in sending_client and
|
|
sending_client.lower() == client.client_type.name.lower()):
|
|
continue
|
|
client.add_event(user_event)
|
|
|
|
def process_event(event, users):
|
|
for user_profile_id in users:
|
|
for client in get_client_descriptors_for_user(user_profile_id):
|
|
if client.accepts_event(event):
|
|
client.add_event(event.copy())
|
|
|
|
def process_userdata_event(event_template, users):
|
|
for user_data in users:
|
|
user_profile_id = user_data['id']
|
|
user_event = event_template.copy() # shallow, but deep enough for our needs
|
|
for key in user_data.keys():
|
|
if key != "id":
|
|
user_event[key] = user_data[key]
|
|
|
|
for client in get_client_descriptors_for_user(user_profile_id):
|
|
if client.accepts_event(user_event):
|
|
client.add_event(user_event)
|
|
|
|
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_message_event(event, users)
|
|
else:
|
|
process_event(event, users)
|
|
|
|
# Runs in the Django process to send a notification to Tornado.
|
|
#
|
|
# We use JSON rather than bare form parameters, so that we can represent
|
|
# different types and for compatibility with non-HTTP transports.
|
|
|
|
def send_notification_http(data):
|
|
if settings.TORNADO_SERVER and not settings.RUNNING_INSIDE_TORNADO:
|
|
requests.post(settings.TORNADO_SERVER + '/notify_tornado', data=dict(
|
|
data = ujson.dumps(data),
|
|
secret = settings.SHARED_SECRET))
|
|
else:
|
|
process_notification(data)
|
|
|
|
def send_notification(data):
|
|
return queue_json_publish("notify_tornado", data, send_notification_http)
|
|
|
|
def send_event(event, users):
|
|
return queue_json_publish("notify_tornado",
|
|
dict(event=event, users=users),
|
|
send_notification_http)
|