nickmbailey/socket_workout.py

## socket_workout.py
"""
socket_workout is a short script designed to reproduce a socket behavior
difference between Jython and CPython.
"""
import socket
import errno
import threading
import traceback
import logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('socket_workout')


class SocketConnectThread(threading.Thread):
    def __init__(self, address, name):
        threading.Thread.__init__(self, name=name)
        self.address = address
        self.connect_status_results = []
        self.final_result = "Test not finished"

    def run(self):
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setblocking(0)

        try:
            connect_errno = 0
            connect_attempt = 0
            while connect_errno != errno.EISCONN and connect_attempt < 100:
                connect_attempt += 1
                connect_errno = sock.connect_ex(self.address)
                self.connect_status_results.append(str(connect_errno))

            if connect_errno == errno.EISCONN:
                self.final_result = "[%s] socket connected - attempts: %s, results = [%s]" % \
                            (self.getName(), str(connect_attempt), ",".join(self.connect_status_results))
            else:
                self.final_result = "[%s] socket failed to connect - attempts: %s, results = [%s]" % \
                            (self.getName(), str(connect_attempt), ",".join(self.connect_status_results))
        except:
            # this shouldn't happen
            self.final_result = traceback.format_exc()
        finally:
            try:
                sock.shutdown(0)
                sock.shutdown(1)
                sock.close()
            except:
                pass


if __name__ == "__main__":
    connect_threads = [SocketConnectThread(address=('127.0.0.1', 8000),
                                           name="SocketTest-" + str(i)) for i in xrange(5)]
    for thread in connect_threads:
        thread.start()
    for thread in connect_threads:
        thread.join()

        logger.debug(thread.final_result)
	"""
	socket_workout is a short script designed to reproduce a socket behavior
	difference between Jython and CPython.
	"""
	import socket
	import errno
	import threading
	import traceback
	import logging
	logging.basicConfig(level=logging.DEBUG)
	logger = logging.getLogger('socket_workout')


	class SocketConnectThread(threading.Thread):
	def __init__(self, address, name):
	threading.Thread.__init__(self, name=name)
	self.address = address
	self.connect_status_results = []
	self.final_result = "Test not finished"

	def run(self):
	sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
	sock.setblocking(0)

	try:
	connect_errno = 0
	connect_attempt = 0
	while connect_errno != errno.EISCONN and connect_attempt < 100:
	connect_attempt += 1
	connect_errno = sock.connect_ex(self.address)
	self.connect_status_results.append(str(connect_errno))

	if connect_errno == errno.EISCONN:
	self.final_result = "[%s] socket connected - attempts: %s, results = [%s]" % \
	(self.getName(), str(connect_attempt), ",".join(self.connect_status_results))
	else:
	self.final_result = "[%s] socket failed to connect - attempts: %s, results = [%s]" % \
	(self.getName(), str(connect_attempt), ",".join(self.connect_status_results))
	except:
	# this shouldn't happen
	self.final_result = traceback.format_exc()
	finally:
	try:
	sock.shutdown(0)
	sock.shutdown(1)
	sock.close()
	except:
	pass


	if __name__ == "__main__":
	connect_threads = [SocketConnectThread(address=('127.0.0.1', 8000),
	name="SocketTest-" + str(i)) for i in xrange(5)]
	for thread in connect_threads:
	thread.start()
	for thread in connect_threads:
	thread.join()

	logger.debug(thread.final_result)