# Python 2
import SocketServer as socketserver
-class ThreadedTCPRequestHandler(socketserver.BaseRequestHandler):
-
- def setup(self):
- if self.server.address_family != socket.AF_UNIX:
- self.request.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
-
+class TestingRequestHandler(socketserver.BaseRequestHandler):
def handle(self):
print("=== socket opened ===")
while True:
elif decoded[0] < 0:
last_eval = decoded
+class ThreadedTCPRequestHandler(TestingRequestHandler):
+ def setup(self):
+ self.request.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+
class ThreadedTCPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
pass
# This requires Python 2.6 or later.
from __future__ import print_function
-from test_channel import ThreadedTCPServer, ThreadedTCPRequestHandler, \
+from test_channel import ThreadedTCPServer, TestingRequestHandler, \
writePortInFile
import socket
import threading
# Python 2
FileNotFoundError = (IOError, OSError)
+if not hasattr(socket, "AF_UNIX"):
+ raise NotImplementedError("Unix sockets are not supported on this platform")
+
class ThreadedUnixServer(ThreadedTCPServer):
address_family = socket.AF_UNIX
+class ThreadedUnixRequestHandler(TestingRequestHandler):
+ pass
+
def main(path):
- server = ThreadedUnixServer(path, ThreadedTCPRequestHandler)
+ server = ThreadedUnixServer(path, ThreadedUnixRequestHandler)
# Start a thread with the server. That thread will then start a new thread
# for each connection.