<s> from __future__ import absolute_import <EOL> from typing import Any <EOL> from django . conf import settings <EOL> from django . utils . timezone import now <EOL> from collections import deque <EOL> import datetime <EOL> import os <EOL> import time <EOL> import socket <EOL> import logging <EOL> import ujson <EOL> import requests <EOL> import atexit <EOL> import sys <EOL> import signal <EOL> import tornado <EOL> import tornado . autoreload <EOL> import random <EOL> import traceback <EOL> from zerver . decorator import RespondAsynchronously , JsonableError <EOL> from zerver . lib . cache import cache_get_many , message_cache_key , user_profile_by_id_cache_key , cache_save_user_profile , cache_with_key <EOL> from zerver . lib . handlers import clear_handler_by_id , get_handler_by_id , finish_handler , handler_stats_string <EOL> from zerver . lib . utils import statsd <EOL> from zerver . middleware import async_request_restart <EOL> from zerver . lib . narrow import build_narrow_filter <EOL> from zerver . lib . queue import queue_json_publish <EOL> from zerver . lib . timestamp import timestamp_to_datetime <EOL> import copy <EOL> import six <EOL> IDLE_EVENT_QUEUE_TIMEOUT_SECS = <NUM_LIT> * <NUM_LIT:10> <EOL> EVENT_QUEUE_GC_FREQ_MSECS = <NUM_LIT:1000> * <NUM_LIT> * <NUM_LIT:5> <EOL> MAX_QUEUE_TIMEOUT_SECS = <NUM_LIT:7> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> <EOL> HEARTBEAT_MIN_FREQ_SECS = <NUM_LIT> <EOL> class ClientDescriptor ( object ) : <EOL> def __init__ ( self , user_profile_id , user_profile_email , realm_id , event_queue , <EOL> event_types , client_type_name , apply_markdown = True , <EOL> all_public_streams = False , lifespan_secs = <NUM_LIT:0> , narrow = [ ] ) : <EOL> self . user_profile_id = user_profile_id <EOL> self . user_profile_email = user_profile_email <EOL> self . realm_id = realm_id <EOL> self . current_handler_id = None <EOL> self . current_client_name = None <EOL> self . event_queue = event_queue <EOL> self . queue_timeout = lifespan_secs <EOL> self . event_types = event_types <EOL> self . last_connection_time = time . time ( ) <EOL> self . apply_markdown = apply_markdown <EOL> self . all_public_streams = all_public_streams <EOL> self . client_type_name = client_type_name <EOL> self . _timeout_handle = None <EOL> self . narrow = narrow <EOL> self . narrow_filter = build_narrow_filter ( narrow ) <EOL> self . queue_timeout = max ( IDLE_EVENT_QUEUE_TIMEOUT_SECS , min ( self . queue_timeout , MAX_QUEUE_TIMEOUT_SECS ) ) <EOL> def to_dict ( self ) : <EOL> return dict ( user_profile_id = self . user_profile_id , <EOL> user_profile_email = self . user_profile_email , <EOL> realm_id = self . realm_id , <EOL> event_queue = self . event_queue . to_dict ( ) , <EOL> queue_timeout = self . queue_timeout , <EOL> event_types = self . event_types , <EOL> last_connection_time = self . last_connection_time , <EOL> apply_markdown = self . apply_markdown , <EOL> all_public_streams = self . all_public_streams , <EOL> narrow = self . narrow , <EOL> client_type_name = self . client_type_name ) <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . event_queue . id , ) <EOL> @ classmethod <EOL> def from_dict ( cls , d ) : <EOL> if '<STR_LIT>' not in d : <EOL> from zerver . models import get_user_profile_by_id <EOL> d [ '<STR_LIT>' ] = get_user_profile_by_id ( d [ '<STR_LIT>' ] ) . email <EOL> if '<STR_LIT>' in d : <EOL> d [ '<STR_LIT>' ] = d [ '<STR_LIT>' ] <EOL> ret = cls ( d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] , <EOL> EventQueue . from_dict ( d [ '<STR_LIT>' ] ) , d [ '<STR_LIT>' ] , <EOL> d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] , <EOL> d [ '<STR_LIT>' ] , d . get ( '<STR_LIT>' , [ ] ) ) <EOL> ret . last_connection_time = d [ '<STR_LIT>' ] <EOL> return ret <EOL> def prepare_for_pickling ( self ) : <EOL> self . current_handler_id = None <EOL> self . _timeout_handle = None <EOL> def add_event ( self , event ) : <EOL> if self . current_handler_id is not None : <EOL> handler = get_handler_by_id ( self . current_handler_id ) <EOL> async_request_restart ( handler . _request ) <EOL> self . event_queue . push ( event ) <EOL> self . finish_current_handler ( ) <EOL> def finish_current_handler ( self , need_timeout = False ) : <EOL> if self . current_handler_id is not None : <EOL> err_msg = "<STR_LIT>" % ( self . event_queue . id , ) <EOL> try : <EOL> finish_handler ( self . current_handler_id , self . event_queue . id , <EOL> self . event_queue . contents ( ) , self . apply_markdown ) <EOL> except Exception : <EOL> logging . exception ( err_msg ) <EOL> finally : <EOL> self . disconnect_handler ( need_timeout = need_timeout ) <EOL> return True <EOL> return False <EOL> def accepts_event ( self , event ) : <EOL> if self . event_types is not None and event [ "<STR_LIT:type>" ] not in self . event_types : <EOL> return False <EOL> if event [ "<STR_LIT:type>" ] == "<STR_LIT:message>" : <EOL> return self . narrow_filter ( event ) <EOL> return True <EOL> def accepts_messages ( self ) : <EOL> return self . event_types is None or "<STR_LIT:message>" in self . event_types <EOL> def idle ( self , now ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . queue_timeout = IDLE_EVENT_QUEUE_TIMEOUT_SECS <EOL> return ( self . current_handler_id is None <EOL> and now - self . last_connection_time >= self . queue_timeout ) <EOL> def connect_handler ( self , handler_id , client_name ) : <EOL> self . current_handler_id = handler_id <EOL> self . current_client_name = client_name <EOL> set_descriptor_by_handler_id ( handler_id , self ) <EOL> self . last_connection_time = time . time ( ) <EOL> def timeout_callback ( ) : <EOL> self . _timeout_handle = None <EOL> self . add_event ( dict ( type = '<STR_LIT>' ) ) <EOL> ioloop = tornado . ioloop . IOLoop . instance ( ) <EOL> heartbeat_time = time . time ( ) + HEARTBEAT_MIN_FREQ_SECS + random . randint ( <NUM_LIT:0> , <NUM_LIT:10> ) <EOL> if self . client_type_name != '<STR_LIT>' : <EOL> self . _timeout_handle = ioloop . add_timeout ( heartbeat_time , timeout_callback ) <EOL> def disconnect_handler ( self , client_closed = False , need_timeout = True ) : <EOL> if self . current_handler_id : <EOL> clear_descriptor_by_handler_id ( self . current_handler_id , None ) <EOL> clear_handler_by_id ( self . current_handler_id ) <EOL> if client_closed : <EOL> logging . info ( "<STR_LIT>" % <EOL> ( self . event_queue . id , self . user_profile_email , <EOL> self . current_client_name ) ) <EOL> self . current_handler_id = None <EOL> self . current_client_name = None <EOL> if need_timeout and self . _timeout_handle is not None : <EOL> ioloop = tornado . ioloop . IOLoop . instance ( ) <EOL> ioloop . remove_timeout ( self . _timeout_handle ) <EOL> self . _timeout_handle = None <EOL> def cleanup ( self ) : <EOL> self . finish_current_handler ( need_timeout = False ) <EOL> do_gc_event_queues ( [ self . event_queue . id ] , [ self . user_profile_id ] , <EOL> [ self . realm_id ] ) <EOL> descriptors_by_handler_id = { } <EOL> def get_descriptor_by_handler_id ( handler_id ) : <EOL> return descriptors_by_handler_id . get ( handler_id ) <EOL> def set_descriptor_by_handler_id ( handler_id , client_descriptor ) : <EOL> descriptors_by_handler_id [ handler_id ] = client_descriptor <EOL> def clear_descriptor_by_handler_id ( handler_id , client_descriptor ) : <EOL> del descriptors_by_handler_id [ handler_id ] <EOL> def compute_full_event_type ( event ) : <EOL> if event [ "<STR_LIT:type>" ] == "<STR_LIT>" : <EOL> if event [ "<STR_LIT:all>" ] : <EOL> return "<STR_LIT>" % ( event [ "<STR_LIT>" ] , event [ "<STR_LIT>" ] ) <EOL> return "<STR_LIT>" % ( event [ "<STR_LIT>" ] , event [ "<STR_LIT>" ] ) <EOL> return event [ "<STR_LIT:type>" ] <EOL> class EventQueue ( object ) : <EOL> def __init__ ( self , id ) : <EOL> self . queue = deque ( ) <EOL> self . next_event_id = <NUM_LIT:0> <EOL> self . id = id <EOL> self . virtual_events = { } <EOL> def to_dict ( self ) : <EOL> return dict ( id = self . id , <EOL> next_event_id = self . next_event_id , <EOL> queue = list ( self . queue ) , <EOL> virtual_events = self . virtual_events ) <EOL> @ classmethod <EOL> def from_dict ( cls , d ) : <EOL> ret = cls ( d [ '<STR_LIT:id>' ] ) <EOL> ret . next_event_id = d [ '<STR_LIT>' ] <EOL> ret . queue = deque ( d [ '<STR_LIT>' ] ) <EOL> ret . virtual_events = d . get ( "<STR_LIT>" , { } ) <EOL> return ret <EOL> def push ( self , event ) : <EOL> event [ '<STR_LIT:id>' ] = self . next_event_id <EOL> self . next_event_id += <NUM_LIT:1> <EOL> full_event_type = compute_full_event_type ( event ) <EOL> if ( full_event_type in [ "<STR_LIT>" , "<STR_LIT>" ] or <EOL> full_event_type . startswith ( "<STR_LIT>" ) ) : <EOL> if full_event_type not in self . virtual_events : <EOL> self . virtual_events [ full_event_type ] = copy . deepcopy ( event ) <EOL> return <EOL> virtual_event = self . virtual_events [ full_event_type ] <EOL> virtual_event [ "<STR_LIT:id>" ] = event [ "<STR_LIT:id>" ] <EOL> if "<STR_LIT>" in event : <EOL> virtual_event [ "<STR_LIT>" ] = event [ "<STR_LIT>" ] <EOL> if full_event_type == "<STR_LIT>" : <EOL> virtual_event [ "<STR_LIT>" ] = event [ "<STR_LIT>" ] <EOL> elif full_event_type == "<STR_LIT>" : <EOL> virtual_event [ "<STR_LIT>" ] = event [ "<STR_LIT>" ] <EOL> elif full_event_type . startswith ( "<STR_LIT>" ) : <EOL> virtual_event [ "<STR_LIT>" ] += event [ "<STR_LIT>" ] <EOL> else : <EOL> self . queue . append ( event ) <EOL> def pop ( self ) : <EOL> return self . queue . popleft ( ) <EOL> def empty ( self ) : <EOL> return len ( self . queue ) == <NUM_LIT:0> and len ( self . virtual_events ) == <NUM_LIT:0> <EOL> def prune ( self , through_id ) : <EOL> while len ( self . queue ) != <NUM_LIT:0> and self . queue [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <= through_id : <EOL> self . pop ( ) <EOL> def contents ( self ) : <EOL> contents = [ ] <EOL> virtual_id_map = { } <EOL> for event_type in self . virtual_events : <EOL> virtual_id_map [ self . virtual_events [ event_type ] [ "<STR_LIT:id>" ] ] = self . virtual_events [ event_type ] <EOL> virtual_ids = sorted ( list ( virtual_id_map . keys ( ) ) ) <EOL> index = <NUM_LIT:0> <EOL> length = len ( virtual_ids ) <EOL> for event in self . queue : <EOL> while index < length and virtual_ids [ index ] < event [ "<STR_LIT:id>" ] : <EOL> contents . append ( virtual_id_map [ virtual_ids [ index ] ] ) <EOL> index += <NUM_LIT:1> <EOL> contents . append ( event ) <EOL> while index < length : <EOL> contents . append ( virtual_id_map [ virtual_ids [ index ] ] ) <EOL> index += <NUM_LIT:1> <EOL> self . virtual_events = { } <EOL> self . queue = deque ( contents ) <EOL> return contents <EOL> clients = { } <EOL> user_clients = { } <EOL> realm_clients_all_streams = { } <EOL> gc_hooks = [ ] <EOL> next_queue_id = <NUM_LIT:0> <EOL> def add_client_gc_hook ( hook ) : <EOL> gc_hooks . append ( hook ) <EOL> def get_client_descriptor ( queue_id ) : <EOL> return clients . get ( queue_id ) <EOL> def get_client_descriptors_for_user ( user_profile_id ) : <EOL> return user_clients . get ( user_profile_id , [ ] ) <EOL> def get_client_descriptors_for_realm_all_streams ( realm_id ) : <EOL> return realm_clients_all_streams . get ( realm_id , [ ] ) <EOL> def add_to_client_dicts ( client ) : <EOL> user_clients . setdefault ( client . user_profile_id , [ ] ) . append ( client ) <EOL> if client . all_public_streams or client . narrow != [ ] : <EOL> realm_clients_all_streams . setdefault ( client . realm_id , [ ] ) . append ( client ) <EOL> def allocate_client_descriptor ( new_queue_data ) : <EOL> global next_queue_id <EOL> queue_id = str ( settings . SERVER_GENERATION ) + '<STR_LIT::>' + str ( next_queue_id ) <EOL> next_queue_id += <NUM_LIT:1> <EOL> new_queue_data [ "<STR_LIT>" ] = EventQueue ( queue_id ) . to_dict ( ) <EOL> client = ClientDescriptor . from_dict ( new_queue_data ) <EOL> clients [ queue_id ] = client <EOL> add_to_client_dicts ( client ) <EOL> return client <EOL> def do_gc_event_queues ( to_remove , affected_users , affected_realms ) : <EOL> def filter_client_dict ( client_dict , key ) : <EOL> if key not in client_dict : <EOL> return <EOL> new_client_list = [ c for c in client_dict [ key ] if c . event_queue . id not in to_remove ] <EOL> if len ( new_client_list ) == <NUM_LIT:0> : <EOL> del client_dict [ key ] <EOL> else : <EOL> client_dict [ key ] = new_client_list <EOL> for user_id in affected_users : <EOL> filter_client_dict ( user_clients , user_id ) <EOL> for realm_id in affected_realms : <EOL> filter_client_dict ( realm_clients_all_streams , realm_id ) <EOL> for id in to_remove : <EOL> for cb in gc_hooks : <EOL> cb ( clients [ id ] . user_profile_id , clients [ id ] , clients [ id ] . user_profile_id not in user_clients ) <EOL> del clients [ id ] <EOL> def gc_event_queues ( ) : <EOL> start = time . time ( ) <EOL> to_remove = set ( ) <EOL> affected_users = set ( ) <EOL> affected_realms = set ( ) <EOL> for ( id , client ) in six . iteritems ( clients ) : <EOL> if client . idle ( start ) : <EOL> to_remove . add ( id ) <EOL> affected_users . add ( client . user_profile_id ) <EOL> affected_realms . add ( client . realm_id ) <EOL> do_gc_event_queues ( to_remove , affected_users , affected_realms ) <EOL> logging . info ( ( '<STR_LIT>' <EOL> + '<STR_LIT>' ) <EOL> % ( len ( to_remove ) , len ( affected_users ) , time . time ( ) - start , <EOL> len ( clients ) , handler_stats_string ( ) ) ) <EOL> statsd . gauge ( '<STR_LIT>' , len ( clients ) ) <EOL> statsd . gauge ( '<STR_LIT>' , len ( user_clients ) ) <EOL> def dump_event_queues ( ) : <EOL> start = time . time ( ) <EOL> with open ( settings . JSON_PERSISTENT_QUEUE_FILENAME , "<STR_LIT:w>" ) as stored_queues : <EOL> ujson . dump ( [ ( qid , client . to_dict ( ) ) for ( qid , client ) in six . iteritems ( clients ) ] , <EOL> stored_queues ) <EOL> logging . info ( '<STR_LIT>' <EOL> % ( len ( clients ) , time . time ( ) - start ) ) <EOL> def load_event_queues ( ) : <EOL> global clients <EOL> start = time . time ( ) <EOL> try : <EOL> with open ( settings . JSON_PERSISTENT_QUEUE_FILENAME , "<STR_LIT:r>" ) as stored_queues : <EOL> json_data = stored_queues . read ( ) <EOL> try : <EOL> clients = dict ( ( qid , ClientDescriptor . from_dict ( client ) ) <EOL> for ( qid , client ) in ujson . loads ( json_data ) ) <EOL> except Exception : <EOL> logging . exception ( "<STR_LIT>" ) <EOL> except ( IOError , EOFError ) : <EOL> pass <EOL> for client in six . itervalues ( clients ) : <EOL> add_to_client_dicts ( client ) <EOL> logging . info ( '<STR_LIT>' <EOL> % ( len ( clients ) , time . time ( ) - start ) ) <EOL> def send_restart_events ( immediate = False ) : <EOL> event = dict ( type = '<STR_LIT>' , server_generation = settings . SERVER_GENERATION ) <EOL> if immediate : <EOL> event [ '<STR_LIT>' ] = True <EOL> for client in six . itervalues ( clients ) : <EOL> if client . accepts_event ( event ) : <EOL> client . add_event ( event . copy ( ) ) <EOL> def setup_event_queue ( ) : <EOL> if not settings . TEST_SUITE : <EOL> load_event_queues ( ) <EOL> atexit . register ( dump_event_queues ) <EOL> signal . signal ( signal . SIGTERM , lambda signum , stack : sys . exit ( <NUM_LIT:1> ) ) <EOL> tornado . autoreload . add_reload_hook ( dump_event_queues ) <EOL> try : <EOL> os . rename ( settings . JSON_PERSISTENT_QUEUE_FILENAME , "<STR_LIT>" ) <EOL> except OSError : <EOL> pass <EOL> ioloop = tornado . ioloop . IOLoop . instance ( ) <EOL> pc = tornado . ioloop . PeriodicCallback ( gc_event_queues , <EOL> EVENT_QUEUE_GC_FREQ_MSECS , ioloop ) <EOL> pc . start ( ) <EOL> send_restart_events ( immediate = settings . DEVELOPMENT ) <EOL> def fetch_events ( query ) : <EOL> queue_id = query [ "<STR_LIT>" ] <EOL> dont_block = query [ "<STR_LIT>" ] <EOL> last_event_id = query [ "<STR_LIT>" ] <EOL> user_profile_id = query [ "<STR_LIT>" ] <EOL> new_queue_data = query . get ( "<STR_LIT>" ) <EOL> user_profile_email = query [ "<STR_LIT>" ] <EOL> client_type_name = query [ "<STR_LIT>" ] <EOL> handler_id = query [ "<STR_LIT>" ] <EOL> try : <EOL> was_connected = False <EOL> orig_queue_id = queue_id <EOL> extra_log_data = "<STR_LIT>" <EOL> if queue_id is None : <EOL> if dont_block : <EOL> client = allocate_client_descriptor ( new_queue_data ) <EOL> queue_id = client . event_queue . id <EOL> else : <EOL> raise JsonableError ( "<STR_LIT>" ) <EOL> else : <EOL> if last_event_id is None : <EOL> raise JsonableError ( "<STR_LIT>" ) <EOL> client = get_client_descriptor ( queue_id ) <EOL> if client is None : <EOL> raise JsonableError ( "<STR_LIT>" % ( queue_id , ) ) <EOL> if user_profile_id != client . user_profile_id : <EOL> raise JsonableError ( "<STR_LIT>" ) <EOL> client . event_queue . prune ( last_event_id ) <EOL> was_connected = client . finish_current_handler ( ) <EOL> if not client . event_queue . empty ( ) or dont_block : <EOL> response = dict ( events = client . event_queue . contents ( ) , <EOL> handler_id = handler_id ) <EOL> if orig_queue_id is None : <EOL> response [ '<STR_LIT>' ] = queue_id <EOL> extra_log_data = "<STR_LIT>" % ( queue_id , len ( response [ "<STR_LIT>" ] ) ) <EOL> if was_connected : <EOL> extra_log_data += "<STR_LIT>" <EOL> return dict ( type = "<STR_LIT>" , response = response , extra_log_data = extra_log_data ) <EOL> if was_connected : <EOL> logging . info ( "<STR_LIT>" % ( queue_id , user_profile_email , <EOL> client_type_name ) ) <EOL> except JsonableError as e : <EOL> if hasattr ( e , '<STR_LIT>' ) and callable ( e . to_json_error_msg ) : <EOL> return dict ( type = "<STR_LIT:error>" , handler_id = handler_id , <EOL> message = e . to_json_error_msg ( ) ) <EOL> raise e <EOL> client . connect_handler ( handler_id , client_type_name ) <EOL> return dict ( type = "<STR_LIT>" ) <EOL> requests_json_is_function = callable ( requests . Response . json ) <EOL> def extract_json_response ( resp ) : <EOL> if requests_json_is_function : <EOL> return resp . json ( ) <EOL> else : <EOL> return resp . json <EOL> def request_event_queue ( user_profile , user_client , apply_markdown , <EOL> queue_lifespan_secs , event_types = None , all_public_streams = False , <EOL> narrow = [ ] ) : <EOL> if settings . TORNADO_SERVER : <EOL> req = { '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : ujson . dumps ( apply_markdown ) , <EOL> '<STR_LIT>' : ujson . dumps ( all_public_streams ) , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : user_client . name , <EOL> '<STR_LIT>' : ujson . dumps ( narrow ) , <EOL> '<STR_LIT>' : queue_lifespan_secs } <EOL> if event_types is not None : <EOL> req [ '<STR_LIT>' ] = ujson . dumps ( event_types ) <EOL> resp = requests . get ( settings . TORNADO_SERVER + '<STR_LIT>' , <EOL> auth = requests . auth . HTTPBasicAuth ( user_profile . email , <EOL> user_profile . api_key ) , <EOL> params = req ) <EOL> resp . raise_for_status ( ) <EOL> return extract_json_response ( resp ) [ '<STR_LIT>' ] <EOL> return None <EOL> def get_user_events ( user_profile , queue_id , last_event_id ) : <EOL> if settings . TORNADO_SERVER : <EOL> resp = requests . get ( settings . TORNADO_SERVER + '<STR_LIT>' , <EOL> auth = requests . auth . HTTPBasicAuth ( user_profile . email , <EOL> user_profile . api_key ) , <EOL> params = { '<STR_LIT>' : queue_id , <EOL> '<STR_LIT>' : last_event_id , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> resp . raise_for_status ( ) <EOL> return extract_json_response ( resp ) [ '<STR_LIT>' ] <EOL> NOTIFY_AFTER_IDLE_HOURS = <NUM_LIT:1> <EOL> def build_offline_notification ( user_profile_id , message_id ) : <EOL> return { "<STR_LIT>" : user_profile_id , <EOL> "<STR_LIT>" : message_id , <EOL> "<STR_LIT>" : time . time ( ) } <EOL> def missedmessage_hook ( user_profile_id , queue , last_for_client ) : <EOL> if not last_for_client : <EOL> return <EOL> message_ids_to_notify = [ ] <EOL> for event in queue . event_queue . contents ( ) : <EOL> if not event [ '<STR_LIT:type>' ] == '<STR_LIT:message>' or not event [ '<STR_LIT>' ] : <EOL> continue <EOL> if '<STR_LIT>' in event [ '<STR_LIT>' ] and not '<STR_LIT>' in event [ '<STR_LIT>' ] : <EOL> notify_info = dict ( message_id = event [ '<STR_LIT:message>' ] [ '<STR_LIT:id>' ] ) <EOL> if not event . get ( '<STR_LIT>' , False ) : <EOL> notify_info [ '<STR_LIT>' ] = True <EOL> if not event . get ( '<STR_LIT>' , False ) : <EOL> notify_info [ '<STR_LIT>' ] = True <EOL> message_ids_to_notify . append ( notify_info ) <EOL> for notify_info in message_ids_to_notify : <EOL> msg_id = notify_info [ '<STR_LIT>' ] <EOL> notice = build_offline_notification ( user_profile_id , msg_id ) <EOL> if notify_info . get ( '<STR_LIT>' , False ) : <EOL> queue_json_publish ( "<STR_LIT>" , notice , lambda notice : None ) <EOL> if notify_info . get ( '<STR_LIT>' , False ) : <EOL> queue_json_publish ( "<STR_LIT>" , notice , lambda notice : None ) <EOL> def receiver_is_idle ( user_profile_id , realm_presences ) : <EOL> all_client_descriptors = get_client_descriptors_for_user ( user_profile_id ) <EOL> message_event_queues = [ client for client in all_client_descriptors if client . accepts_messages ( ) ] <EOL> off_zulip = len ( message_event_queues ) == <NUM_LIT:0> <EOL> if realm_presences is None or not user_profile_id in realm_presences : <EOL> return off_zulip <EOL> user_presence = realm_presences [ user_profile_id ] <EOL> latest_active_timestamp = None <EOL> idle = False <EOL> for client , status in six . iteritems ( user_presence ) : <EOL> if ( latest_active_timestamp is None or status [ '<STR_LIT>' ] > latest_active_timestamp ) and status [ '<STR_LIT:status>' ] == '<STR_LIT>' : <EOL> latest_active_timestamp = status [ '<STR_LIT>' ] <EOL> if latest_active_timestamp is None : <EOL> idle = True <EOL> else : <EOL> active_datetime = timestamp_to_datetime ( latest_active_timestamp ) <EOL> idle = now ( ) - active_datetime > datetime . timedelta ( seconds = <NUM_LIT> ) <EOL> return off_zulip or idle <EOL> def process_message_event ( event_template , users ) : <EOL> realm_presences = { int ( k ) : v for k , v in event_template [ '<STR_LIT>' ] . items ( ) } <EOL> sender_queue_id = event_template . get ( '<STR_LIT>' , None ) <EOL> message_dict_markdown = event_template [ '<STR_LIT>' ] <EOL> message_dict_no_markdown = event_template [ '<STR_LIT>' ] <EOL> sender_id = message_dict_markdown [ '<STR_LIT>' ] <EOL> message_id = message_dict_markdown [ '<STR_LIT:id>' ] <EOL> message_type = message_dict_markdown [ '<STR_LIT:type>' ] <EOL> sending_client = message_dict_markdown [ '<STR_LIT>' ] <EOL> send_to_clients = dict ( ) <EOL> extra_user_data = { } <EOL> if '<STR_LIT>' in event_template and not event_template . get ( "<STR_LIT>" ) : <EOL> for client in get_client_descriptors_for_realm_all_streams ( event_template [ '<STR_LIT>' ] ) : <EOL> send_to_clients [ client . event_queue . id ] = { '<STR_LIT>' : client , '<STR_LIT>' : None } <EOL> if sender_queue_id is not None and client . event_queue . id == sender_queue_id : <EOL> send_to_clients [ client . event_queue . id ] [ '<STR_LIT>' ] = True <EOL> for user_data in users : <EOL> user_profile_id = user_data [ '<STR_LIT:id>' ] <EOL> flags = user_data . get ( '<STR_LIT>' , [ ] ) <EOL> for client in get_client_descriptors_for_user ( user_profile_id ) : <EOL> send_to_clients [ client . event_queue . id ] = { '<STR_LIT>' : client , '<STR_LIT>' : flags } <EOL> if sender_queue_id is not None and client . event_queue . id == sender_queue_id : <EOL> send_to_clients [ client . event_queue . id ] [ '<STR_LIT>' ] = True <EOL> received_pm = message_type == "<STR_LIT>" and user_profile_id != sender_id <EOL> mentioned = '<STR_LIT>' in flags <EOL> idle = receiver_is_idle ( user_profile_id , realm_presences ) <EOL> always_push_notify = user_data . get ( '<STR_LIT>' , False ) <EOL> if ( received_pm or mentioned ) and ( idle or always_push_notify ) : <EOL> notice = build_offline_notification ( user_profile_id , message_id ) <EOL> queue_json_publish ( "<STR_LIT>" , notice , lambda notice : None ) <EOL> notified = dict ( push_notified = True ) <EOL> if idle : <EOL> queue_json_publish ( "<STR_LIT>" , notice , lambda notice : None ) <EOL> notified [ '<STR_LIT>' ] = True <EOL> extra_user_data [ user_profile_id ] = notified <EOL> for client_data in six . itervalues ( send_to_clients ) : <EOL> client = client_data [ '<STR_LIT>' ] <EOL> flags = client_data [ '<STR_LIT>' ] <EOL> is_sender = client_data . get ( '<STR_LIT>' , False ) <EOL> extra_data = extra_user_data . get ( client . user_profile_id , None ) <EOL> if not client . accepts_messages ( ) : <EOL> continue <EOL> if client . apply_markdown : <EOL> message_dict = message_dict_markdown <EOL> else : <EOL> message_dict = message_dict_no_markdown <EOL> if "<STR_LIT>" in client . client_type_name and event_template . get ( "<STR_LIT>" ) : <EOL> message_dict = message_dict . copy ( ) <EOL> message_dict [ "<STR_LIT>" ] = True <EOL> user_event = dict ( type = '<STR_LIT:message>' , message = message_dict , flags = flags ) <EOL> if extra_data is not None : <EOL> user_event . update ( extra_data ) <EOL> if is_sender : <EOL> local_message_id = event_template . get ( '<STR_LIT>' , None ) <EOL> if local_message_id is not None : <EOL> user_event [ "<STR_LIT>" ] = local_message_id <EOL> if not client . accepts_event ( user_event ) : <EOL> continue <EOL> if ( '<STR_LIT>' in sending_client and <EOL> sending_client . lower ( ) == client . client_type_name . lower ( ) ) : <EOL> continue <EOL> client . add_event ( user_event ) <EOL> def process_event ( event , users ) : <EOL> for user_profile_id in users : <EOL> for client in get_client_descriptors_for_user ( user_profile_id ) : <EOL> if client . accepts_event ( event ) : <EOL> client . add_event ( event . copy ( ) ) <EOL> def process_userdata_event ( event_template , users ) : <EOL> for user_data in users : <EOL> user_profile_id = user_data [ '<STR_LIT:id>' ] <EOL> user_event = event_template . copy ( ) <EOL> for key in user_data . keys ( ) : <EOL> if key != "<STR_LIT:id>" : <EOL> user_event [ key ] = user_data [ key ] <EOL> for client in get_client_descriptors_for_user ( user_profile_id ) : <EOL> if client . accepts_event ( user_event ) : <EOL> client . add_event ( user_event ) <EOL> def process_notification ( notice ) : <EOL> event = notice [ '<STR_LIT>' ] <EOL> users = notice [ '<STR_LIT>' ] <EOL> if event [ '<STR_LIT:type>' ] in [ "<STR_LIT>" ] : <EOL> process_userdata_event ( event , users ) <EOL> elif event [ '<STR_LIT:type>' ] == "<STR_LIT:message>" : <EOL> process_message_event ( event , users ) <EOL> else : <EOL> process_event ( event , users ) <EOL> def send_notification_http ( data ) : <EOL> if settings . TORNADO_SERVER and not settings . RUNNING_INSIDE_TORNADO : <EOL> requests . post ( settings . TORNADO_SERVER + '<STR_LIT>' , data = dict ( <EOL> data = ujson . dumps ( data ) , <EOL> secret = settings . SHARED_SECRET ) ) <EOL> else : <EOL> process_notification ( data ) <EOL> def send_notification ( data ) : <EOL> return queue_json_publish ( "<STR_LIT>" , data , send_notification_http ) <EOL> def send_event ( event , users ) : <EOL> return queue_json_publish ( "<STR_LIT>" , <EOL> dict ( event = event , users = users ) , <EOL> send_notification_http ) </s>
