home · contact · privacy
Minor code rearrangements.
[plomrogue2-experiments] / server_ / io.py
1 import socketserver
2 import threading
3 import queue
4
5
6 # Avoid "Address already in use" errors.
7 socketserver.TCPServer.allow_reuse_address = True
8
9
10 # Our default server port.
11 SERVER_PORT=5000
12
13
14 class Server(socketserver.ThreadingTCPServer):
15     """Bind together threaded IO handling server and message queue."""
16
17     def __init__(self, queue, *args, **kwargs):
18         super().__init__(('localhost', SERVER_PORT), IO_Handler, *args, **kwargs)
19         self.queue_out = queue
20         self.daemon_threads = True  # Else, server's threads have daemon=False.
21
22
23 class IO_Handler(socketserver.BaseRequestHandler):
24
25     def handle(self):
26         """Move messages between network socket and game IO loop via queues.
27
28         On start (a new connection from client to server), sets up a
29         new queue, sends it via self.server.queue_out to the game IO
30         loop thread, and from then on receives messages to send back
31         from the game IO loop via that new queue.
32
33         At the same time, loops over socket's recv to get messages
34         from the outside into the game IO loop by way of
35         self.server.queue_out into the game IO. Ends connection once a
36         'QUIT' message is received from socket, and then also calls
37         for a kill of its own queue.
38
39         All messages to the game IO loop are tuples, with the first
40         element a meta command ('ADD_QUEUE' for queue creation,
41         'KILL_QUEUE' for queue deletion, and 'COMMAND' for everything
42         else), the second element a UUID that uniquely identifies the
43         thread (so that the game IO loop knows whom to send replies
44         back to), and optionally a third element for further
45         instructions.
46
47         """
48         import plom_socket_io
49
50         def caught_send(socket, message):
51             """Send message by socket, catch broken socket connection error."""
52             try:
53                 plom_socket_io.send(socket, message)
54             except plom_socket_io.BrokenSocketConnection:
55                 pass
56
57         def send_queue_messages(socket, queue_in, thread_alive):
58             """Send messages via socket from queue_in while thread_alive[0]."""
59             while thread_alive[0]:
60                 try:
61                     msg = queue_in.get(timeout=1)
62                 except queue.Empty:
63                     continue
64                 caught_send(socket, msg)
65
66         import uuid
67         print('CONNECTION FROM:', str(self.client_address))
68         connection_id = uuid.uuid4()
69         queue_in = queue.Queue()
70         self.server.queue_out.put(('ADD_QUEUE', connection_id, queue_in))
71         thread_alive = [True]
72         t = threading.Thread(target=send_queue_messages,
73                              args=(self.request, queue_in, thread_alive))
74         t.start()
75         for message in plom_socket_io.recv(self.request):
76             if message is None:
77                 caught_send(self.request, 'BAD MESSAGE')
78             elif 'QUIT' == message:
79                 caught_send(self.request, 'BYE')
80                 break
81             else:
82                 self.server.queue_out.put(('COMMAND', connection_id, message))
83         self.server.queue_out.put(('KILL_QUEUE', connection_id))
84         thread_alive[0] = False
85         print('CONNECTION CLOSED FROM:', str(self.client_address))
86         self.request.close()
87
88
89 def io_loop(q, game_command_handler):
90     """Handle commands coming through queue q, send results back.
91
92     Commands from q are expected to be tuples, with the first element
93     either 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element
94     a UUID, and an optional third element of arbitrary type. The UUID
95     identifies a receiver for replies.
96
97     An 'ADD_QUEUE' command should contain as third element a queue
98     through which to send messages back to the sender of the
99     command. A 'KILL_QUEUE' command removes the queue for that
100     receiver from the list of queues through which to send replies.
101
102     A 'COMMAND' command is specified in greater detail by a string
103     that is the tuple's third element. The game_command_handler takes
104     care of processing this and sending out replies.
105
106     """
107     while True:
108         x = q.get()
109         command_type = x[0]
110         connection_id = x[1]
111         content = None if len(x) == 2 else x[2]
112         if command_type == 'ADD_QUEUE':
113             game_command_handler.queues_out[connection_id] = content
114         elif command_type == 'KILL_QUEUE':
115             del game_command_handler.queues_out[connection_id]
116         elif command_type == 'COMMAND':
117             game_command_handler.handle_input(content, connection_id)
118
119
120 def run_server_with_io_loop(command_handler):
121     """Run connection of server talking to clients and game IO loop.
122
123     We have the TCP server (an instance of Server) and we have the
124     game IO loop, a thread running io_loop. Both communicate with each
125     other via a queue.Queue. While the TCP server may spawn parallel
126     threads to many clients, the IO loop works sequentially through
127     game commands received from the TCP server's threads (= client
128     connections to the TCP server), calling command_handler to process
129     them. A processed command may trigger messages to the commanding
130     client or to all clients, delivered from the IO loop to the TCP
131     server via the queue.
132
133     """
134     q = queue.Queue()
135     c = threading.Thread(target=io_loop, daemon=True, args=(q, command_handler))
136     c.start()
137     server = Server(q)
138     try:
139         server.serve_forever()
140     except KeyboardInterrupt:
141         pass
142     finally:
143         print('Killing server')
144         server.server_close()