<s> import struct <EOL> from twisted . internet import defer <EOL> from twisted . internet import protocol <EOL> from twimp import chunks <EOL> from twimp import const <EOL> from twimp . crypto . handshake import CryptoHandshaker <EOL> from twimp . dispatch import EventDispatchFactory , CallDispatchProtocol <EOL> from twimp . error import CallResultError , CallAbortedException <EOL> from twimp . error import ConnectFailedError , InvalidAppError <EOL> from twimp . error import PlayFailed , PlayNotFound <EOL> from twimp . primitives import _s_ulong_b as _s_ulong <EOL> from twimp . proto import UserControlDispatchDemuxer <EOL> from twimp . server import errors <EOL> from twimp . helpers import vb <EOL> LOG_CATEGORY = '<STR_LIT>' <EOL> import twimp . log <EOL> log = twimp . log . get_logger ( LOG_CATEGORY ) <EOL> _s_uc_single = struct . Struct ( '<STR_LIT>' ) <EOL> class NetStream ( object ) : <EOL> def __init__ ( self , protocol , stream_id ) : <EOL> self . protocol = protocol <EOL> self . id = stream_id <EOL> self . buffer_length = <NUM_LIT:100> <EOL> self . protocol . route_buffer_messages ( self . id , self . _set_buffer_length ) <EOL> def _set_buffer_length ( self , ts , length ) : <EOL> self . buffer_length = length <EOL> def close ( self ) : <EOL> self . unset_listeners ( ) <EOL> self . protocol . route_buffer_messages ( self . id , None ) <EOL> def set_listeners ( self , data_callback = None , meta_callback = None , <EOL> mute_callback = None ) : <EOL> p = self . protocol <EOL> p . route_data_messages ( self . id , data_callback ) <EOL> p . route_meta_messages ( self . id , meta_callback ) <EOL> p . route_mute_messages ( self . id , mute_callback ) <EOL> def unset_listeners ( self ) : <EOL> p = self . protocol <EOL> for method in ( p . route_data_messages , p . route_meta_messages , <EOL> p . route_mute_messages ) : <EOL> method ( self . id , None ) <EOL> def send ( self , ts , type_ , data ) : <EOL> return self . protocol . muxer . sendMessage ( ts , type_ , self . id , data ) <EOL> def asend ( self , ts , type_ , * args ) : <EOL> return self . protocol . muxer . sendMessage ( ts , type_ , self . id , <EOL> self . protocol . encode_amf ( * args ) ) <EOL> def call ( self , cmd , * args , ** kwargs ) : <EOL> return self . protocol . callRemote ( self . id , cmd , * args , ** kwargs ) <EOL> def signal ( self , cmd , * args , ** kwargs ) : <EOL> return self . protocol . signalRemote ( self . id , cmd , * args , ** kwargs ) <EOL> def ctrlStreamBegin ( self ) : <EOL> sm = self . protocol . muxer . sendMessage <EOL> return sm ( <NUM_LIT:0> , chunks . PROTO_USER_CONTROL , <NUM_LIT:0> , <EOL> vb ( _s_uc_single . pack ( const . UCTRL_STREAM_BEGIN , self . id ) ) ) <EOL> def ctrlStreamRecorded ( self ) : <EOL> sm = self . protocol . muxer . sendMessage <EOL> return sm ( <NUM_LIT:0> , chunks . PROTO_USER_CONTROL , <NUM_LIT:0> , <EOL> vb ( _s_uc_single . pack ( const . UCTRL_STREAM_RECORDED , self . id ) ) ) <EOL> def set_chunk_size ( self , new_size ) : <EOL> sm = self . protocol . muxer . sendMessage <EOL> sm ( <NUM_LIT:0> , chunks . PROTO_SET_CHUNK_SIZE , <NUM_LIT:0> , vb ( _s_ulong . pack ( new_size ) ) ) <EOL> self . protocol . muxer . set_chunk_size ( new_size ) <EOL> class NetStreamManager ( object ) : <EOL> def __init__ ( self ) : <EOL> self . _streams = { } <EOL> self . _next_id = <NUM_LIT:1> <EOL> def make_stream ( self , protocol ) : <EOL> s_id , self . _next_id = self . _next_id , self . _next_id + <NUM_LIT:1> <EOL> s = NetStream ( protocol , s_id ) <EOL> self . _streams [ s_id ] = s <EOL> return s <EOL> def del_stream ( self , ms_id ) : <EOL> self . _streams . pop ( ms_id , None ) <EOL> def get_stream ( self , ms_id ) : <EOL> return self . _streams . get ( ms_id , None ) <EOL> def get_streams ( self ) : <EOL> return self . _streams . values ( ) <EOL> class DelegateSelectedUserControlDemuxer ( UserControlDispatchDemuxer ) : <EOL> def doUserControlBufferLength ( self , header , stream_id , length ) : <EOL> self . protocol . demuxerUCBufferLength ( header . abs_time , header . ms_id , <EOL> stream_id , length ) <EOL> def check_connected ( method ) : <EOL> def wrapped ( self , * args , ** kwargs ) : <EOL> if not self . _connected : <EOL> self . _fail ( '<STR_LIT>' ) <EOL> return <EOL> return method ( self , * args , ** kwargs ) <EOL> wrapped . __name__ = method . __name__ <EOL> return wrapped <EOL> def check_connected_remote ( method ) : <EOL> def wrapped ( self , * args , ** kwargs ) : <EOL> if not self . _connected : <EOL> self . _fail ( '<STR_LIT>' ) <EOL> raise CallAbortedException ( '<STR_LIT>' ) <EOL> return method ( self , * args , ** kwargs ) <EOL> wrapped . __name__ = method . __name__ <EOL> return wrapped <EOL> def _harden_connect ( method ) : <EOL> def wrapped ( self , * args , ** kwargs ) : <EOL> try : <EOL> return method ( self , * args , ** kwargs ) <EOL> except ( CallResultError , CallAbortedException ) : <EOL> raise <EOL> except Exception , e : <EOL> raise ConnectFailedError ( repr ( e ) , fatal = True ) <EOL> wrapped . __name__ = method . __name__ <EOL> return wrapped <EOL> rtmp_to_internal_types = { <EOL> const . RTMP_AUDIO : chunks . MSG_AUDIO , <EOL> const . RTMP_VIDEO : chunks . MSG_VIDEO , <EOL> } <EOL> CDP = CallDispatchProtocol <EOL> class AppDispatchServerProtocol ( CDP ) : <EOL> handshaker_class = CryptoHandshaker <EOL> demuxer_class = DelegateSelectedUserControlDemuxer <EOL> def __init__ ( self ) : <EOL> CDP . __init__ ( self ) <EOL> self . _app = None <EOL> self . _nsmgr = NetStreamManager ( ) <EOL> self . _connected = False <EOL> self . _data_routes = { } <EOL> self . _meta_routes = { } <EOL> self . _mute_routes = { } <EOL> self . _buffer_routes = { } <EOL> def handshakeSucceeded ( self , init_ts , hs_delay ) : <EOL> CDP . handshakeSucceeded ( self , init_ts , hs_delay ) <EOL> self . transport . setTcpNoDelay ( <NUM_LIT:1> ) <EOL> def connectionLost ( self , reason = protocol . connectionDone ) : <EOL> if self . _app : <EOL> self . _app . connectionLost ( reason ) <EOL> self . _app = None <EOL> for ns in self . _nsmgr . get_streams ( ) : <EOL> ns . close ( ) <EOL> CDP . connectionLost ( self , reason ) <EOL> def _fail ( self , msg = None ) : <EOL> if msg is not None : <EOL> log . info ( '<STR_LIT>' , msg ) <EOL> self . transport . loseConnection ( ) <EOL> @ check_connected_remote <EOL> def unknownRemoteCall ( self , cmd , ts , ms_id , args ) : <EOL> handler_m = getattr ( self . _app , '<STR_LIT>' % ( cmd , ) , None ) <EOL> if ms_id == <NUM_LIT:0> or handler_m is None : <EOL> return CDP . unknownRemoteCall ( self , cmd , ts , ms_id , args ) <EOL> ns = self . _nsmgr . get_stream ( ms_id ) <EOL> if not ns : <EOL> raise CallResultError ( '<STR_LIT>' % ( ms_id , ) ) <EOL> return handler_m ( ts , ns , * args ) <EOL> @ _harden_connect <EOL> def remote_connect ( self , ts , ms_id , cmd_obj , * opts ) : <EOL> app_path = cmd_obj . app <EOL> if app_path is None : <EOL> raise InvalidAppError ( '<STR_LIT>' , fatal = True ) <EOL> d = None <EOL> app_factory_args = self . factory . get_app_factory ( app_path ) <EOL> if app_factory_args : <EOL> app_factory , path_args , path_kwargs = app_factory_args <EOL> self . _app = app_factory ( self , self . factory . server ) <EOL> d = defer . maybeDeferred ( self . _app . connect , cmd_obj , opts , <EOL> * path_args , ** path_kwargs ) <EOL> d . addCallbacks ( self . _connect_succeeded , self . _connect_failed ) <EOL> else : <EOL> raise InvalidAppError ( '<STR_LIT>' , fatal = True ) <EOL> return d <EOL> def _connect_succeeded ( self , result ) : <EOL> self . _connected = True <EOL> sm = self . muxer . sendMessage <EOL> sm ( <NUM_LIT:0> , chunks . PROTO_WINDOW_SIZE , <NUM_LIT:0> , vb ( '<STR_LIT>' . decode ( '<STR_LIT>' ) ) ) <EOL> sm ( <NUM_LIT:0> , chunks . PROTO_SET_BANDWIDTH , <NUM_LIT:0> , vb ( '<STR_LIT>' . decode ( '<STR_LIT>' ) ) ) <EOL> sm ( <NUM_LIT:0> , chunks . PROTO_USER_CONTROL , <NUM_LIT:0> , vb ( '<STR_LIT>' . decode ( '<STR_LIT>' ) ) ) <EOL> return result <EOL> def _connect_failed ( self , failure ) : <EOL> if not failure . check ( CallResultError , CallAbortedException ) : <EOL> raise ConnectFailedError ( repr ( failure . value ) , fatal = True ) <EOL> return failure <EOL> @ check_connected_remote <EOL> def remote_createStream ( self , ts , ms_id , _none ) : <EOL> s = self . _nsmgr . make_stream ( self ) <EOL> log . info ( '<STR_LIT>' , s . id ) <EOL> return None , s . id <EOL> @ check_connected_remote <EOL> def remote_play ( self , ts , ms_id , * args ) : <EOL> return self . _call_play ( ms_id , args ) <EOL> @ check_connected_remote <EOL> def remote_publish ( self , ts , ms_id , * args ) : <EOL> return self . _call_publish ( ms_id , args ) <EOL> def _call_play ( self , ms_id , args ) : <EOL> ns = self . _nsmgr . get_stream ( ms_id ) <EOL> if not ns : <EOL> raise PlayFailed ( '<STR_LIT>' % ( ms_id , ) ) <EOL> def translate_not_found_eb ( failure ) : <EOL> failure . trap ( errors . NotFoundError ) <EOL> raise PlayNotFound ( str ( failure . value ) ) <EOL> def play_cb ( result ) : <EOL> log . debug ( '<STR_LIT>' ) <EOL> return result <EOL> d = defer . maybeDeferred ( self . _app . play , ns , * args [ <NUM_LIT:1> : ] ) <EOL> d . addCallbacks ( play_cb , translate_not_found_eb ) <EOL> return d <EOL> def _call_publish ( self , ms_id , args ) : <EOL> ns = self . _nsmgr . get_stream ( ms_id ) <EOL> if not ns : <EOL> raise CallResultError ( '<STR_LIT>' % ( ms_id , ) ) <EOL> d = defer . maybeDeferred ( self . _app . publish , ns , * args [ <NUM_LIT:1> : ] ) <EOL> return d <EOL> def _set_route_messages ( self , ms_id , callback , table ) : <EOL> if callback : <EOL> table [ ms_id ] = callback <EOL> else : <EOL> table . pop ( ms_id , None ) <EOL> def route_data_messages ( self , ms_id , callback ) : <EOL> self . _set_route_messages ( ms_id , callback , self . _data_routes ) <EOL> def route_meta_messages ( self , ms_id , callback ) : <EOL> self . _set_route_messages ( ms_id , callback , self . _meta_routes ) <EOL> def route_mute_messages ( self , ms_id , callback ) : <EOL> self . _set_route_messages ( ms_id , callback , self . _mute_routes ) <EOL> def route_buffer_messages ( self , ms_id , callback ) : <EOL> self . _set_route_messages ( ms_id , callback , self . _buffer_routes ) <EOL> def doMeta ( self , ts , ms_id , args ) : <EOL> handler = self . _meta_routes . get ( ms_id ) <EOL> if handler : <EOL> handler ( ts , args ) <EOL> elif not self . _connected : <EOL> self . _fail ( '<STR_LIT>' ) <EOL> return <EOL> def doData ( self , type_ , ts , ms_id , body ) : <EOL> msg_type = rtmp_to_internal_types [ type_ ] <EOL> handler = self . _data_routes . get ( ms_id ) <EOL> if handler : <EOL> handler ( ts , msg_type , body ) <EOL> elif not self . _connected : <EOL> self . _fail ( '<STR_LIT>' ) <EOL> return <EOL> @ check_connected_remote <EOL> def remote_receiveAudio ( self , ts , ms_id , _none , do_receive ) : <EOL> handler = self . _mute_routes . get ( ms_id ) <EOL> if handler : <EOL> handler ( ts , chunks . MSG_AUDIO , do_receive ) <EOL> @ check_connected_remote <EOL> def remote_receiveVideo ( self , ts , ms_id , _none , do_receive ) : <EOL> handler = self . _mute_routes . get ( ms_id ) <EOL> if handler : <EOL> handler ( ts , chunks . MSG_VIDEO , do_receive ) <EOL> @ check_connected <EOL> def demuxerUCBufferLength ( self , ts , ms_id , stream_id , length ) : <EOL> handler = self . _buffer_routes . get ( stream_id ) <EOL> if handler : <EOL> handler ( ts , length ) <EOL> class AppDispatchServerFactory ( EventDispatchFactory ) : <EOL> protocol = AppDispatchServerProtocol <EOL> server = None <EOL> def __init__ ( self , init_time , server ) : <EOL> EventDispatchFactory . __init__ ( self , init_time ) <EOL> self . server = server <EOL> def get_app_factory ( self , app_path ) : <EOL> """<STR_LIT>""" <EOL> class URLDispatchingServerFactory ( AppDispatchServerFactory ) : <EOL> def __init__ ( self , init_time , server , urls ) : <EOL> AppDispatchServerFactory . __init__ ( self , init_time , server ) <EOL> self . urls = urls <EOL> def get_app_factory ( self , app_path ) : <EOL> for pattern , app_factory in self . urls : <EOL> m = pattern . match ( app_path ) <EOL> if m : <EOL> kwargs = m . groupdict ( ) <EOL> if kwargs : <EOL> args = ( ) <EOL> else : <EOL> args = m . groups ( ) <EOL> return app_factory , args , kwargs <EOL> return None <EOL> def make_urls ( * urls ) : <EOL> import re <EOL> return [ ( re . compile ( pattern ) , factory ) for ( pattern , factory ) in urls ] </s>
