The code below fails. Socket.accept blocks inside the thread and doesn't
let go, even after the socket was closed. From the error presented, the
socket never actually closes.
I realize that the below is a basic Java idiom but how do you do the
same thing in Python?
import unittest
import socket
import threading
import time
class SocketAcceptor ( threading.Thread ):
def __init__( self, socket ):
threading.Thread.__init__( self )
self.socket = socket
def run( self ):
self.socket.bind( ( "", 3423 ) )
self.socket.listen( 5 )
child, ip = self.socket.accept()
class SocketTester ( unittest.TestCase ):
def testClose( self ):
for each in range( 4 ):
ss = socket.socket()
acceptor_thread = SocketAcceptor( ss )
acceptor_thread.start()
time.sleep( 1 )
ss.close()
if __name__ == '__main__':
unittest.main()
