+class Server(socketserver.ThreadingTCPServer):
+ """Bind together threaded IO handling server and message queue."""
+
+ def __init__(self, queue, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ self.queue_out = queue
+ self.daemon_threads = True # Else, server's threads have daemon=False.
+
+
+class IO_Handler(socketserver.BaseRequestHandler):
+
+ def handle(self):
+ """Move messages between network socket and main thread via queues.
+
+ On start, sets up new queue, sends it via self.server.queue_out to
+ main thread, and from then on receives messages to send back from the
+ main thread via that new queue.
+
+ At the same time, loops over socket's recv to get messages from the
+ outside via self.server.queue_out into the main thread. Ends connection
+ once a 'QUIT' message is received from socket, and then also kills its
+ own queue.
+
+ All messages to the main thread are tuples, with the first element a
+ meta command ('ADD_QUEUE' for queue creation, 'KILL_QUEUE' for queue
+ deletion, and 'COMMAND' for everything else), the second element a UUID
+ that uniquely identifies the thread (so that the main thread knows whom
+ to send replies back to), and optionally a third element for further
+ instructions.
+ """
+ import plom_socket_io
+
+ def caught_send(socket, message):
+ """Send message by socket, catch broken socket connection error."""
+ try:
+ plom_socket_io.send(socket, message)
+ except plom_socket_io.BrokenSocketConnection:
+ pass
+
+ def send_queue_messages(socket, queue_in, thread_alive):
+ """Send messages via socket from queue_in while thread_alive[0]."""
+ while thread_alive[0]:
+ try:
+ msg = queue_in.get(timeout=1)
+ except queue.Empty:
+ continue
+ caught_send(socket, msg)
+
+ import uuid
+ print('CONNECTION FROM:', str(self.client_address))
+ connection_id = uuid.uuid4()
+ queue_in = queue.Queue()
+ self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
+ thread_alive = [True]
+ t = threading.Thread(target=send_queue_messages,
+ args=(self.request, queue_in, thread_alive))
+ t.start()
+ for message in plom_socket_io.recv(self.request):
+ if message is None:
+ caught_send(self.request, 'BAD MESSAGE')
+ elif 'QUIT' == message:
+ caught_send(self.request, 'BYE')
+ break
+ else:
+ self.server.queue_out.put(('COMMAND', connection_id, message))
+ self.server.queue_out.put(('KILL_QUEUE', connection_id))
+ thread_alive[0] = False
+ print('CONNECTION CLOSED FROM:', str(self.client_address))
+ self.request.close()
+
+