<s> import socket <EOL> from threading import Lock , Thread <EOL> from PyQt5 . QtCore import QObject , pyqtSlot , pyqtSignal <EOL> class SocketServer ( QObject , Thread ) : <EOL> onopen = pyqtSignal ( ) <EOL> onconnect = pyqtSignal ( QObject ) <EOL> onmessage = pyqtSignal ( QObject , str ) <EOL> ondisconnect = pyqtSignal ( QObject ) <EOL> onclose = pyqtSignal ( ) <EOL> onerror = pyqtSignal ( int , str ) <EOL> def __init__ ( self , parent , encoding = '<STR_LIT:utf-8>' , timeout = None ) : <EOL> super ( SocketServer , self ) . __init__ ( parent ) <EOL> self . setObjectName ( '<STR_LIT>' ) <EOL> self . setDaemon ( True ) <EOL> self . encoding = encoding <EOL> self . timeout = timeout <EOL> self . socket = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> def run ( self ) : <EOL> lock = Lock ( ) <EOL> while True : <EOL> try : <EOL> lock . acquire ( ) <EOL> conn , addr = self . socket . accept ( ) <EOL> socketConn = SocketConn ( self , conn , self . encoding , self . timeout ) <EOL> self . onconnect . emit ( socketConn ) <EOL> lock . release ( ) <EOL> except Exception as e : <EOL> print ( e ) <EOL> break <EOL> try : <EOL> self . onclose . emit ( ) <EOL> except : <EOL> pass <EOL> @ pyqtSlot ( int , result = bool ) <EOL> def listen ( self , port ) : <EOL> try : <EOL> self . socket . bind ( ( '<STR_LIT>' , port ) ) <EOL> self . socket . listen ( <NUM_LIT:10> ) <EOL> self . start ( ) <EOL> self . onopen . emit ( ) <EOL> return True <EOL> except Exception as e : <EOL> self . onerror . emit ( e . errno , e . strerror ) <EOL> return False <EOL> @ pyqtSlot ( ) <EOL> def close ( self ) : <EOL> self . socket . close ( ) <EOL> def messageHandler ( self , socketConn , message ) : <EOL> self . onmessage . emit ( socketConn , message ) <EOL> def disconnectHandler ( self , socketConn ) : <EOL> self . ondisconnect . emit ( socketConn ) <EOL> class SocketConn ( QObject , Thread ) : <EOL> onmessage = pyqtSignal ( str ) <EOL> onclose = pyqtSignal ( ) <EOL> onerror = pyqtSignal ( int , str ) <EOL> def __init__ ( self , parent , conn , encoding = '<STR_LIT:utf-8>' , timeout = None ) : <EOL> super ( SocketConn , self ) . __init__ ( ) <EOL> self . setObjectName ( '<STR_LIT>' ) <EOL> self . parent = parent <EOL> self . conn = conn <EOL> self . encoding = encoding <EOL> self . conn . settimeout ( timeout ) <EOL> self . setDaemon ( True ) <EOL> self . start ( ) <EOL> def run ( self ) : <EOL> lock = Lock ( ) <EOL> while True : <EOL> try : <EOL> lock . acquire ( ) <EOL> bytes = self . conn . recv ( <NUM_LIT> ) <EOL> if bytes : <EOL> self . messageHandler ( bytes ) <EOL> else : <EOL> break <EOL> lock . release ( ) <EOL> except Exception as e : <EOL> print ( e ) <EOL> break <EOL> try : <EOL> self . close ( ) <EOL> self . onclose . emit ( ) <EOL> self . parent . disconnectHandler ( self ) <EOL> except : <EOL> pass <EOL> @ pyqtSlot ( str , result = int ) <EOL> def send ( self , message ) : <EOL> try : <EOL> bytes = bytearray ( message , self . encoding ) <EOL> return self . conn . send ( bytes ) <EOL> except : <EOL> return <NUM_LIT:0> <EOL> @ pyqtSlot ( ) <EOL> def close ( self ) : <EOL> self . conn . close ( ) <EOL> def messageHandler ( self , bytes ) : <EOL> message = str ( bytes , self . encoding ) <EOL> self . onmessage . emit ( message ) <EOL> self . parent . messageHandler ( self , message ) </s>
