<s> import socket , traceback , threading , queue <EOL> from . log import log <EOL> class Interop : <EOL> def __init__ ( self , on_recv , on_connect , on_not_connected ) : <EOL> self . readWorker = None <EOL> self . readWorkerStopEvent = None <EOL> self . readBuffer = bytes ( ) <EOL> self . socket = None <EOL> self . on_connect = on_connect <EOL> self . on_recv = on_recv <EOL> self . on_not_connected = on_not_connected <EOL> self . sendWorker = None <EOL> self . sendQueue = queue . Queue ( ) <EOL> self . sendDataEvent = None <EOL> self . sendWorkerStopEvent = None <EOL> def isConnected ( self ) : <EOL> isConnected = self . socket != None <EOL> return isConnected <EOL> def connect ( self ) : <EOL> try : <EOL> host = "<STR_LIT:localhost>" <EOL> port = <NUM_LIT> <EOL> log ( ) . info ( "<STR_LIT>" % ( host , port ) ) <EOL> tmpSocket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> tmpSocket . connect ( ( host , port ) ) <EOL> except OSError : <EOL> log ( ) . info ( "<STR_LIT>" % ( host , port ) ) <EOL> tmpSocket . close ( ) <EOL> return <EOL> self . socket = tmpSocket <EOL> self . startPullMessages ( ) <EOL> self . startSendMessages ( ) <EOL> self . on_connect ( ) <EOL> log ( ) . info ( "<STR_LIT>" ) <EOL> def send ( self , type , msg ) : <EOL> if not self . isConnected ( ) : <EOL> self . on_not_connected ( ) <EOL> if not self . isConnected ( ) : <EOL> return <EOL> self . sendQueue . put ( ( type , msg ) , True ) <EOL> self . sendDataEvent . set ( ) <EOL> def startSendMessages ( self ) : <EOL> self . sendDataEvent = threading . Event ( ) <EOL> self . sendWorkerStopEvent = threading . Event ( ) <EOL> self . sendWorker = threading . Thread ( target = self . sendMessages ) <EOL> self . sendWorker . daemon = True <EOL> self . sendWorker . start ( ) <EOL> def sendMessages ( self ) : <EOL> try : <EOL> while not self . sendWorkerStopEvent . is_set ( ) : <EOL> while not self . sendQueue . empty ( ) : <EOL> msgTupple = self . sendQueue . get_nowait ( ) <EOL> type = msgTupple [ <NUM_LIT:0> ] <EOL> msg = msgTupple [ <NUM_LIT:1> ] <EOL> msgInBytes = bytes ( type + "<STR_LIT:\n>" + str ( len ( msg ) ) + "<STR_LIT:\n>" + msg , "<STR_LIT>" ) <EOL> self . socket . sendall ( msgInBytes ) <EOL> self . sendDataEvent . wait ( ) <EOL> self . sendDataEvent . clear ( ) <EOL> except : <EOL> self . disconnect ( ) <EOL> def stopSendMessages ( self ) : <EOL> self . sendWorkerStopEvent . set ( ) <EOL> self . sendDataEvent . set ( ) <EOL> def startPullMessages ( self ) : <EOL> self . readWorkerStopEvent = threading . Event ( ) <EOL> self . readWorker = threading . Thread ( target = self . pullMessages ) <EOL> self . readWorker . daemon = True <EOL> self . readWorker . start ( ) <EOL> def stopPullMessages ( self ) : <EOL> self . readWorkerStopEvent . set ( ) <EOL> def pullMessages ( self ) : <EOL> try : <EOL> while not self . readWorkerStopEvent . is_set ( ) : <EOL> tmpData = self . socket . recv ( <NUM_LIT> ) <EOL> if len ( tmpData ) == <NUM_LIT:0> : <EOL> log ( ) . info ( "<STR_LIT>" ) <EOL> self . disconnect ( ) <EOL> return <EOL> self . readBuffer = self . readBuffer + tmpData <EOL> self . parseReadData ( ) <EOL> except : <EOL> self . disconnect ( ) <EOL> return <EOL> def parseReadData ( self ) : <EOL> strData = self . readBuffer . decode ( "<STR_LIT:utf-8>" ) <EOL> firstNewLine = strData . find ( "<STR_LIT:\n>" ) <EOL> secondNewLine = strData . find ( "<STR_LIT:\n>" , firstNewLine + <NUM_LIT:1> ) <EOL> if firstNewLine <= <NUM_LIT:0> or secondNewLine <= <NUM_LIT:0> : <EOL> return <EOL> typeStr = strData [ : firstNewLine ] <EOL> lengthStr = strData [ firstNewLine + <NUM_LIT:1> : secondNewLine ] <EOL> length = self . parseLength ( lengthStr ) <EOL> if length == - <NUM_LIT:1> : <EOL> self . readBuffer = b"<STR_LIT>" ; <EOL> return <EOL> sizeOfTypeStr = len ( bytes ( typeStr , "<STR_LIT:utf-8>" ) ) + <NUM_LIT:1> <EOL> sizeOfLengthStr = len ( bytes ( lengthStr , "<STR_LIT:utf-8>" ) ) + <NUM_LIT:1> <EOL> if len ( self . readBuffer ) - sizeOfLengthStr - sizeOfTypeStr < length : <EOL> return <EOL> tmpStr = self . readBuffer [ sizeOfLengthStr + sizeOfTypeStr : length + sizeOfLengthStr + sizeOfTypeStr ] <EOL> message = ( typeStr , tmpStr . decode ( "<STR_LIT:utf-8>" ) ) <EOL> self . readBuffer = self . readBuffer [ sizeOfLengthStr + length + sizeOfTypeStr : ] <EOL> self . on_recv ( message ) <EOL> self . parseReadData ( ) <EOL> def parseLength ( self , lenStr ) : <EOL> try : <EOL> return int ( lenStr ) <EOL> except ValueError : <EOL> log ( ) . info ( "<STR_LIT>" + lenStr ) <EOL> return - <NUM_LIT:1> <EOL> def disconnect ( self ) : <EOL> log ( ) . info ( "<STR_LIT>" ) <EOL> if self . readWorkerStopEvent != None : <EOL> self . stopPullMessages ( ) <EOL> if self . sendWorkerStopEvent != None : <EOL> self . stopSendMessages ( ) <EOL> try : <EOL> if self . socket != None : <EOL> self . socket . shutdown ( socket . SHUT_RDWR ) <EOL> self . socket . close ( ) <EOL> except : <EOL> pass <EOL> finally : <EOL> self . socket = None <EOL> self . readWorkerStopEvent = None <EOL> self . readWorker = None <EOL> self . readBuffer = bytes ( ) <EOL> self . sendWorkerStopEvent = None <EOL> self . sendWorker = None <EOL> self . sendDataEvent = None <EOL> log ( ) . info ( "<STR_LIT>" ) </s>
