<s> MAX_ACK_QUEUE = <NUM_LIT> <EOL> MAX_MSG_QUEUE = <NUM_LIT> <EOL> import thread <EOL> import time <EOL> from ant . core . constants import * <EOL> from ant . core . message import Message , ChannelEventMessage <EOL> from ant . core . exceptions import MessageError <EOL> def ProcessBuffer ( buffer_ ) : <EOL> messages = [ ] <EOL> while True : <EOL> hf = Message ( ) <EOL> try : <EOL> msg = hf . getHandler ( buffer_ ) <EOL> buffer_ = buffer_ [ len ( msg . getPayload ( ) ) + <NUM_LIT:4> : ] <EOL> messages . append ( msg ) <EOL> except MessageError , e : <EOL> if e . internal == "<STR_LIT>" : <EOL> buffer_ = buffer_ [ ord ( buffer_ [ <NUM_LIT:1> ] ) + <NUM_LIT:4> : ] <EOL> else : <EOL> break <EOL> return ( buffer_ , messages , ) <EOL> def EventPump ( evm ) : <EOL> evm . pump_lock . acquire ( ) <EOL> evm . pump = True <EOL> evm . pump_lock . release ( ) <EOL> go = True <EOL> buffer_ = '<STR_LIT>' <EOL> while go : <EOL> evm . running_lock . acquire ( ) <EOL> if not evm . running : <EOL> go = False <EOL> evm . running_lock . release ( ) <EOL> buffer_ += evm . driver . read ( <NUM_LIT:20> ) <EOL> if len ( buffer_ ) == <NUM_LIT:0> : <EOL> continue <EOL> buffer_ , messages = ProcessBuffer ( buffer_ ) <EOL> evm . callbacks_lock . acquire ( ) <EOL> for message in messages : <EOL> for callback in evm . callbacks : <EOL> try : <EOL> callback . process ( message ) <EOL> except Exception , e : <EOL> pass <EOL> evm . callbacks_lock . release ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> evm . pump_lock . acquire ( ) <EOL> evm . pump = False <EOL> evm . pump_lock . release ( ) <EOL> class EventCallback ( object ) : <EOL> def process ( self , msg ) : <EOL> pass <EOL> class AckCallback ( EventCallback ) : <EOL> def __init__ ( self , evm ) : <EOL> self . evm = evm <EOL> def process ( self , msg ) : <EOL> if isinstance ( msg , ChannelEventMessage ) : <EOL> self . evm . ack_lock . acquire ( ) <EOL> self . evm . ack . append ( msg ) <EOL> if len ( self . evm . ack ) > MAX_ACK_QUEUE : <EOL> self . evm . ack = self . evm . ack [ - MAX_ACK_QUEUE : ] <EOL> self . evm . ack_lock . release ( ) <EOL> class MsgCallback ( EventCallback ) : <EOL> def __init__ ( self , evm ) : <EOL> self . evm = evm <EOL> def process ( self , msg ) : <EOL> self . evm . msg_lock . acquire ( ) <EOL> self . evm . msg . append ( msg ) <EOL> if len ( self . evm . msg ) > MAX_MSG_QUEUE : <EOL> self . evm . msg = self . evm . msg [ - MAX_MSG_QUEUE : ] <EOL> self . evm . msg_lock . release ( ) <EOL> class EventMachine ( object ) : <EOL> callbacks_lock = thread . allocate_lock ( ) <EOL> running_lock = thread . allocate_lock ( ) <EOL> pump_lock = thread . allocate_lock ( ) <EOL> ack_lock = thread . allocate_lock ( ) <EOL> msg_lock = thread . allocate_lock ( ) <EOL> def __init__ ( self , driver ) : <EOL> self . driver = driver <EOL> self . callbacks = [ ] <EOL> self . running = False <EOL> self . pump = False <EOL> self . ack = [ ] <EOL> self . msg = [ ] <EOL> self . registerCallback ( AckCallback ( self ) ) <EOL> self . registerCallback ( MsgCallback ( self ) ) <EOL> def registerCallback ( self , callback ) : <EOL> self . callbacks_lock . acquire ( ) <EOL> if callback not in self . callbacks : <EOL> self . callbacks . append ( callback ) <EOL> self . callbacks_lock . release ( ) <EOL> def removeCallback ( self , callback ) : <EOL> self . callbacks_lock . acquire ( ) <EOL> if callback in self . callbacks : <EOL> self . callbacks . remove ( callback ) <EOL> self . callbacks_lock . release ( ) <EOL> def waitForAck ( self , msg ) : <EOL> while True : <EOL> self . ack_lock . acquire ( ) <EOL> for emsg in self . ack : <EOL> if msg . getType ( ) != emsg . getMessageID ( ) : <EOL> continue <EOL> self . ack . remove ( emsg ) <EOL> self . ack_lock . release ( ) <EOL> return emsg . getMessageCode ( ) <EOL> self . ack_lock . release ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> def waitForMessage ( self , class_ ) : <EOL> while True : <EOL> self . msg_lock . acquire ( ) <EOL> for emsg in self . msg : <EOL> if not isinstance ( emsg , class_ ) : <EOL> continue <EOL> self . msg . remove ( emsg ) <EOL> self . msg_lock . release ( ) <EOL> return emsg <EOL> self . msg_lock . release ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> def start ( self , driver = None ) : <EOL> self . running_lock . acquire ( ) <EOL> if self . running : <EOL> self . running_lock . release ( ) <EOL> return <EOL> self . running = True <EOL> if driver is not None : <EOL> self . driver = driver <EOL> thread . start_new_thread ( EventPump , ( self , ) ) <EOL> while True : <EOL> self . pump_lock . acquire ( ) <EOL> if self . pump : <EOL> self . pump_lock . release ( ) <EOL> break <EOL> self . pump_lock . release ( ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> self . running_lock . release ( ) <EOL> def stop ( self ) : <EOL> self . running_lock . acquire ( ) <EOL> if not self . running : <EOL> self . running_lock . release ( ) <EOL> return <EOL> self . running = False <EOL> self . running_lock . release ( ) <EOL> while True : <EOL> self . pump_lock . acquire ( ) <EOL> if not self . pump : <EOL> self . pump_lock . release ( ) <EOL> break <EOL> self . pump_lock . release ( ) <EOL> time . sleep ( <NUM_LIT> ) </s>
