#NEWLINE# def __call__( self , environ , start_response ) : #NEWLINE# #INDENT# '\n        Hijack the main loop from the original thread and listen on events on the Redis\n        and the Websocket filedescriptors.\n        ' #NEWLINE# websocket = None #NEWLINE# subscriber = self . Subscriber ( self . _redis_connection ) #NEWLINE# try : #NEWLINE# #INDENT# self . assure_protocol_requirements ( environ ) #NEWLINE# request = WSGIRequest ( environ ) #NEWLINE# if callable ( private_settings . WS4REDIS_PROCESS_REQUEST ) : #NEWLINE# #INDENT# private_settings . WS4REDIS_PROCESS_REQUEST ( request ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# self . process_request ( request ) #NEWLINE# #UNINDENT# ( channels , echo_message ) = self . process_subscriptions ( request ) #NEWLINE# if callable ( private_settings . WS4REDIS_ALLOWED_CHANNELS ) : #NEWLINE# #INDENT# channels = list ( private_settings . WS4REDIS_ALLOWED_CHANNELS ( request , channels ) ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# if ( private_settings . WS4REDIS_ALLOWED_CHANNELS is not None ) : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# ( mod , callback ) = private_settings . WS4REDIS_ALLOWED_CHANNELS . rsplit ( '.' , 1 ) #NEWLINE# callback = getattr ( import_module ( mod ) , callback , None ) #NEWLINE# if callable ( callback ) : #NEWLINE# #INDENT# channels = list ( callback ( request , channels ) ) #NEWLINE# #UNINDENT# #UNINDENT# except AttributeError : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# websocket = self . upgrade_websocket ( environ , start_response ) #NEWLINE# logger . debug ( 'Subscribed to channels: {0}' . format ( ', ' . join ( channels ) ) ) #NEWLINE# subscriber . set_pubsub_channels ( request , channels ) #NEWLINE# websocket_fd = websocket . get_file_descriptor ( ) #NEWLINE# listening_fds = [ websocket_fd ] #NEWLINE# redis_fd = subscriber . get_file_descriptor ( ) #NEWLINE# if redis_fd : #NEWLINE# #INDENT# listening_fds . append ( redis_fd ) #NEWLINE# #UNINDENT# subscriber . send_persited_messages ( websocket ) #NEWLINE# recvmsg = None #NEWLINE# while ( websocket and ( not websocket . closed ) ) : #NEWLINE# #INDENT# ready = self . select ( listening_fds , [ ] , [ ] , 4.0 ) [ 0 ] #NEWLINE# if ( not ready ) : #NEWLINE# #INDENT# websocket . flush ( ) #NEWLINE# #UNINDENT# for fd in ready : #NEWLINE# #INDENT# if ( fd == websocket_fd ) : #NEWLINE# #INDENT# recvmsg = RedisMessage ( websocket . receive ( ) ) #NEWLINE# if recvmsg : #NEWLINE# #INDENT# subscriber . publish_message ( recvmsg ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# if ( fd == redis_fd ) : #NEWLINE# #INDENT# sendmsg = RedisMessage ( subscriber . parse_response ( ) ) #NEWLINE# if ( sendmsg and ( echo_message or ( sendmsg != recvmsg ) ) ) : #NEWLINE# #INDENT# websocket . send ( sendmsg ) #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# logger . error ( 'Invalid file descriptor: {0}' . format ( fd ) ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# if ( private_settings . WS4REDIS_HEARTBEAT and ( not websocket . closed ) ) : #NEWLINE# #INDENT# websocket . send ( private_settings . WS4REDIS_HEARTBEAT ) #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# except WebSocketError as excpt : #NEWLINE# #INDENT# logger . warning ( 'WebSocketError: {}' . format ( excpt ) , exc_info = sys . exc_info ( ) ) #NEWLINE# response = http . HttpResponse ( status = 1001 , content = 'Websocket Closed' ) #NEWLINE# #UNINDENT# except UpgradeRequiredError as excpt : #NEWLINE# #INDENT# logger . info ( 'Websocket upgrade required' ) #NEWLINE# response = http . HttpResponseBadRequest ( status = 426 , content = excpt ) #NEWLINE# #UNINDENT# except HandshakeError as excpt : #NEWLINE# #INDENT# logger . warning ( 'HandshakeError: {}' . format ( excpt ) , exc_info = sys . exc_info ( ) ) #NEWLINE# response = http . HttpResponseBadRequest ( content = excpt ) #NEWLINE# #UNINDENT# except PermissionDenied as excpt : #NEWLINE# #INDENT# logger . warning ( 'PermissionDenied: {}' . format ( excpt ) , exc_info = sys . exc_info ( ) ) #NEWLINE# response = http . HttpResponseForbidden ( content = excpt ) #NEWLINE# #UNINDENT# except Exception as excpt : #NEWLINE# #INDENT# logger . error ( 'Other Exception: {}' . format ( excpt ) , exc_info = sys . exc_info ( ) ) #NEWLINE# response = http . HttpResponseServerError ( content = excpt ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# response = http . HttpResponse ( ) #NEWLINE# #UNINDENT# finally : #NEWLINE# #INDENT# subscriber . release ( ) #NEWLINE# if websocket : #NEWLINE# #INDENT# websocket . close ( code = 1001 , message = 'Websocket Closed' ) #NEWLINE# #UNINDENT# else : #NEWLINE# #INDENT# logger . warning ( 'Starting late response on websocket' ) #NEWLINE# status_text = http_client . responses . get ( response . status_code , 'UNKNOWN STATUS CODE' ) #NEWLINE# status = '{0} {1}' . format ( response . status_code , status_text ) #NEWLINE# headers = response . _headers . values ( ) #NEWLINE# if six . PY3 : #NEWLINE# #INDENT# headers = list ( headers ) #NEWLINE# #UNINDENT# start_response ( force_str ( status ) , headers ) #NEWLINE# logger . info ( 'Finish non-websocket response with status code: {}' . format ( response . status_code ) ) #NEWLINE# #UNINDENT# #UNINDENT# return response