<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import , print_function <EOL> import atexit <EOL> import collections <EOL> import inspect <EOL> import itertools <EOL> import sys <EOL> import traceback <EOL> from six . moves import range as irange <EOL> import ejson <EOL> import gevent <EOL> import geventwebsocket <EOL> from django . conf import settings <EOL> from django . core import signals <EOL> from django . core . handlers . base import BaseHandler <EOL> from django . core . handlers . wsgi import WSGIRequest <EOL> from django . db import connection , transaction <EOL> from dddp import alea , this , ADDED , CHANGED , REMOVED , MeteorError <EOL> def safe_call ( func , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return None , func ( * args , ** kwargs ) <EOL> except Exception : <EOL> return traceback . format_exc ( ) , None <EOL> def dprint ( name , val ) : <EOL> """<STR_LIT>""" <EOL> from pprint import pformat <EOL> print ( <EOL> '<STR_LIT>' % ( <EOL> name , <EOL> '<STR_LIT>' . join ( <EOL> pformat ( <EOL> val , indent = <NUM_LIT:4> , width = <NUM_LIT> , <EOL> ) . split ( '<STR_LIT:\n>' ) <EOL> ) , <EOL> ) , <EOL> ) <EOL> def validate_kwargs ( func , kwargs ) : <EOL> """<STR_LIT>""" <EOL> func_name = func . __name__ <EOL> argspec = inspect . getargspec ( func ) <EOL> all_args = argspec . args [ : ] <EOL> defaults = list ( argspec . defaults or [ ] ) <EOL> if inspect . ismethod ( func ) and all_args [ : <NUM_LIT:1> ] == [ '<STR_LIT>' ] : <EOL> all_args [ : <NUM_LIT:1> ] = [ ] <EOL> if defaults : <EOL> required = all_args [ : - len ( defaults ) ] <EOL> else : <EOL> required = all_args [ : ] <EOL> trans = { <EOL> arg : arg . endswith ( '<STR_LIT:_>' ) and arg [ : - <NUM_LIT:1> ] or arg <EOL> for arg <EOL> in all_args <EOL> } <EOL> for key in list ( kwargs ) : <EOL> key_adj = '<STR_LIT>' % key <EOL> if key_adj in all_args : <EOL> kwargs [ key_adj ] = kwargs . pop ( key ) <EOL> supplied = sorted ( kwargs ) <EOL> missing = [ <EOL> trans . get ( arg , arg ) for arg in required <EOL> if arg not in supplied <EOL> ] <EOL> if missing : <EOL> raise MeteorError ( <EOL> <NUM_LIT> , <EOL> func . err , <EOL> '<STR_LIT>' % ( <EOL> func_name , <EOL> '<STR_LIT:U+0020>' . join ( missing ) , <EOL> ) , <EOL> ) <EOL> extra = [ <EOL> arg for arg in supplied <EOL> if arg not in all_args <EOL> ] <EOL> if extra : <EOL> raise MeteorError ( <EOL> <NUM_LIT> , <EOL> func . err , <EOL> '<STR_LIT>' % ( func_name , '<STR_LIT:U+0020>' . join ( extra ) ) , <EOL> ) <EOL> class DDPWebSocketApplication ( geventwebsocket . WebSocketApplication ) : <EOL> """<STR_LIT>""" <EOL> _tx_buffer = None <EOL> _tx_buffer_id_gen = None <EOL> _tx_next_id_gen = None <EOL> _tx_next_id = None <EOL> methods = { } <EOL> versions = [ <EOL> '<STR_LIT:1>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> api = None <EOL> logger = None <EOL> pgworker = None <EOL> remote_addr = None <EOL> version = None <EOL> support = None <EOL> connection = None <EOL> remote_ids = None <EOL> base_handler = BaseHandler ( ) <EOL> def get_tx_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return next ( self . _tx_buffer_id_gen ) <EOL> def on_open ( self ) : <EOL> """<STR_LIT>""" <EOL> this . request = WSGIRequest ( self . ws . environ ) <EOL> this . ws = self <EOL> this . send = self . send <EOL> this . reply = self . reply <EOL> self . logger = self . ws . logger <EOL> self . remote_ids = collections . defaultdict ( set ) <EOL> self . _tx_buffer = { } <EOL> self . _tx_buffer_id_gen = itertools . cycle ( irange ( sys . maxint ) ) <EOL> self . _tx_next_id_gen = itertools . cycle ( irange ( sys . maxint ) ) <EOL> self . _tx_next_id = next ( self . _tx_next_id_gen ) <EOL> this . remote_addr = self . remote_addr = '<STR_LIT>' . format ( <EOL> self . ws . environ , <EOL> ) <EOL> this . subs = { } <EOL> safe_call ( self . logger . info , '<STR_LIT>' , self ) <EOL> self . send ( '<STR_LIT:o>' ) <EOL> self . send ( '<STR_LIT>' ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . remote_addr <EOL> def on_close ( self , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . connection is not None : <EOL> del self . pgworker . connections [ self . connection . pk ] <EOL> self . connection . delete ( ) <EOL> self . connection = None <EOL> signals . request_finished . send ( sender = self . __class__ ) <EOL> safe_call ( self . logger . info , '<STR_LIT>' , self , args or '<STR_LIT>' ) <EOL> def on_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> if self . ws . closed : <EOL> return None <EOL> try : <EOL> safe_call ( self . logger . debug , '<STR_LIT>' , self , message ) <EOL> for data in self . ddp_frames_from_message ( message ) : <EOL> self . process_ddp ( data ) <EOL> signals . request_finished . send ( sender = self . __class__ ) <EOL> except geventwebsocket . WebSocketError : <EOL> self . ws . close ( ) <EOL> def ddp_frames_from_message ( self , message ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> msgs = ejson . loads ( message ) <EOL> except ValueError : <EOL> self . reply ( <EOL> '<STR_LIT:error>' , error = <NUM_LIT> , reason = '<STR_LIT>' , <EOL> ) <EOL> raise StopIteration <EOL> if not isinstance ( msgs , list ) : <EOL> self . reply ( <EOL> '<STR_LIT:error>' , error = <NUM_LIT> , reason = '<STR_LIT>' , <EOL> ) <EOL> raise StopIteration <EOL> while msgs : <EOL> raw = msgs . pop ( <NUM_LIT:0> ) <EOL> try : <EOL> data = ejson . loads ( raw ) <EOL> except ( TypeError , ValueError ) : <EOL> data = None <EOL> if not isinstance ( data , dict ) : <EOL> self . reply ( <EOL> '<STR_LIT:error>' , error = <NUM_LIT> , <EOL> reason = '<STR_LIT>' , <EOL> offendingMessage = raw , <EOL> ) <EOL> yield data <EOL> if msgs : <EOL> gevent . sleep ( ) <EOL> def process_ddp ( self , data ) : <EOL> """<STR_LIT>""" <EOL> msg_id = data . get ( '<STR_LIT:id>' , None ) <EOL> try : <EOL> msg = data . pop ( '<STR_LIT>' ) <EOL> except KeyError : <EOL> self . reply ( <EOL> '<STR_LIT:error>' , reason = '<STR_LIT>' , <EOL> offendingMessage = data , <EOL> ) <EOL> return <EOL> try : <EOL> self . dispatch ( msg , data ) <EOL> except Exception as err : <EOL> kwargs = { <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT:result>' } . get ( msg , '<STR_LIT:error>' ) , <EOL> } <EOL> if msg_id is not None : <EOL> kwargs [ '<STR_LIT:id>' ] = msg_id <EOL> if isinstance ( err , MeteorError ) : <EOL> error = err . as_dict ( ) <EOL> else : <EOL> error = { <EOL> '<STR_LIT:error>' : <NUM_LIT> , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> if kwargs [ '<STR_LIT>' ] == '<STR_LIT:error>' : <EOL> kwargs . update ( error ) <EOL> else : <EOL> kwargs [ '<STR_LIT:error>' ] = error <EOL> if not isinstance ( err , MeteorError ) : <EOL> stack , _ = safe_call ( <EOL> self . logger . error , '<STR_LIT>' , msg , data , exc_info = <NUM_LIT:1> , <EOL> ) <EOL> if stack is not None : <EOL> traceback . print_exc ( file = sys . stderr ) <EOL> sys . stderr . write ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> sys . stderr . write ( stack ) <EOL> elif settings . DEBUG : <EOL> print ( '<STR_LIT>' % err ) <EOL> dprint ( '<STR_LIT>' , msg ) <EOL> dprint ( '<STR_LIT:data>' , data ) <EOL> error . setdefault ( '<STR_LIT>' , traceback . format_exc ( ) ) <EOL> print ( error [ '<STR_LIT>' ] ) <EOL> self . reply ( ** kwargs ) <EOL> if msg_id and msg == '<STR_LIT>' : <EOL> self . reply ( '<STR_LIT>' , methods = [ msg_id ] ) <EOL> @ transaction . atomic <EOL> def dispatch ( self , msg , kwargs ) : <EOL> """<STR_LIT>""" <EOL> if self . connection is None and msg != '<STR_LIT>' : <EOL> self . reply ( '<STR_LIT:error>' , reason = '<STR_LIT>' ) <EOL> return <EOL> if msg == '<STR_LIT>' : <EOL> if ( <EOL> '<STR_LIT>' not in kwargs <EOL> ) or ( <EOL> '<STR_LIT:id>' not in kwargs <EOL> ) : <EOL> self . reply ( <EOL> '<STR_LIT:error>' , error = <NUM_LIT> , reason = '<STR_LIT>' , <EOL> ) <EOL> return <EOL> try : <EOL> handler = getattr ( self , '<STR_LIT>' % msg ) <EOL> except ( AttributeError , UnicodeEncodeError ) : <EOL> raise MeteorError ( <NUM_LIT> , '<STR_LIT>' ) <EOL> validate_kwargs ( handler , kwargs ) <EOL> handler ( ** kwargs ) <EOL> def send ( self , data , tx_id = None ) : <EOL> """<STR_LIT>""" <EOL> if tx_id is None : <EOL> tx_id = self . get_tx_id ( ) <EOL> self . _tx_buffer [ tx_id ] = data <EOL> while self . _tx_next_id in self . _tx_buffer : <EOL> data = self . _tx_buffer . pop ( self . _tx_next_id ) <EOL> if self . _tx_buffer : <EOL> safe_call ( self . logger . debug , '<STR_LIT>' , self . _tx_next_id ) <EOL> self . _tx_next_id = next ( self . _tx_next_id_gen ) <EOL> if not isinstance ( data , basestring ) : <EOL> msg = data . get ( '<STR_LIT>' , None ) <EOL> if msg in ( ADDED , CHANGED , REMOVED ) : <EOL> ids = self . remote_ids [ data [ '<STR_LIT>' ] ] <EOL> meteor_id = data [ '<STR_LIT:id>' ] <EOL> if msg == ADDED : <EOL> if meteor_id in ids : <EOL> msg = data [ '<STR_LIT>' ] = CHANGED <EOL> else : <EOL> ids . add ( meteor_id ) <EOL> elif msg == CHANGED : <EOL> if meteor_id not in ids : <EOL> msg = data [ '<STR_LIT>' ] = ADDED <EOL> ids . add ( meteor_id ) <EOL> elif msg == REMOVED : <EOL> try : <EOL> ids . remove ( meteor_id ) <EOL> except KeyError : <EOL> continue <EOL> data = '<STR_LIT>' % ejson . dumps ( [ ejson . dumps ( data ) ] ) <EOL> safe_call ( self . logger . debug , '<STR_LIT>' , self , data ) <EOL> try : <EOL> self . ws . send ( data ) <EOL> except geventwebsocket . WebSocketError : <EOL> self . ws . close ( ) <EOL> self . _tx_buffer . clear ( ) <EOL> break <EOL> num_waiting = len ( self . _tx_buffer ) <EOL> if num_waiting > <NUM_LIT:10> : <EOL> safe_call ( <EOL> self . logger . warn , <EOL> '<STR_LIT>' , <EOL> tx_id , self . _tx_next_id , num_waiting , self . _tx_buffer , <EOL> ) <EOL> def reply ( self , msg , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs [ '<STR_LIT>' ] = msg <EOL> self . send ( kwargs ) <EOL> def recv_connect ( self , version = None , support = None , session = None ) : <EOL> """<STR_LIT>""" <EOL> del session <EOL> if self . connection is not None : <EOL> raise MeteorError ( <EOL> <NUM_LIT> , '<STR_LIT>' , <EOL> self . connection . connection_id , <EOL> ) <EOL> elif None in ( version , support ) or version not in self . versions : <EOL> self . reply ( '<STR_LIT>' , version = self . versions [ <NUM_LIT:0> ] ) <EOL> elif version not in support : <EOL> raise MeteorError ( <NUM_LIT> , '<STR_LIT>' ) <EOL> else : <EOL> from dddp . models import Connection <EOL> cur = connection . cursor ( ) <EOL> cur . execute ( '<STR_LIT>' ) <EOL> ( backend_pid , ) = cur . fetchone ( ) <EOL> this . version = version <EOL> this . support = support <EOL> self . connection = Connection . objects . create ( <EOL> server_addr = '<STR_LIT>' % ( <EOL> backend_pid , <EOL> self . ws . handler . socket . getsockname ( ) , <EOL> ) , <EOL> remote_addr = self . remote_addr , <EOL> version = version , <EOL> ) <EOL> self . pgworker . connections [ self . connection . pk ] = self <EOL> atexit . register ( self . on_close , '<STR_LIT>' ) <EOL> self . reply ( '<STR_LIT>' , session = self . connection . connection_id ) <EOL> recv_connect . err = '<STR_LIT>' <EOL> def recv_ping ( self , id_ = None ) : <EOL> """<STR_LIT>""" <EOL> if id_ is None : <EOL> self . reply ( '<STR_LIT>' ) <EOL> else : <EOL> self . reply ( '<STR_LIT>' , id = id_ ) <EOL> recv_ping . err = '<STR_LIT>' <EOL> def recv_sub ( self , id_ , name , params ) : <EOL> """<STR_LIT>""" <EOL> self . api . sub ( id_ , name , * params ) <EOL> recv_sub . err = '<STR_LIT>' <EOL> def recv_unsub ( self , id_ = None ) : <EOL> """<STR_LIT>""" <EOL> if id_ : <EOL> self . api . unsub ( id_ ) <EOL> else : <EOL> self . reply ( '<STR_LIT>' ) <EOL> recv_unsub . err = '<STR_LIT>' <EOL> def recv_method ( self , method , params , id_ , randomSeed = None ) : <EOL> """<STR_LIT>""" <EOL> if randomSeed is not None : <EOL> this . random_streams . random_seed = randomSeed <EOL> this . alea_random = alea . Alea ( randomSeed ) <EOL> self . api . method ( method , params , id_ ) <EOL> self . reply ( '<STR_LIT>' , methods = [ id_ ] ) <EOL> recv_method . err = '<STR_LIT>' </s>
