plomlompom.com Git - plomrogue2-experiments/blob - server_/io.py

   1 import socketserver
   2 import threading
   3 import queue
   4 import sys
   5 sys.path.append('../')
   6 import parser
   7 from server_.game_error import GameError
   8
   9
  10 # Avoid "Address already in use" errors.
  11 socketserver.TCPServer.allow_reuse_address = True
  12
  13
  14 class Server(socketserver.ThreadingTCPServer):
  15     """Bind together threaded IO handling server and message queue."""
  16
  17     def __init__(self, queue, port, *args, **kwargs):
  18         super().__init__(('localhost', 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 class GameIO():
  90
  91     def __init__(self, game_file_name, game):
  92         self.game_file_name = game_file_name
  93         self.queues_out = {}
  94         self.parser = parser.Parser(game)
  95
  96     def loop(self, q):
  97         """Handle commands coming through queue q, send results back.
  98
  99         Commands from q are expected to be tuples, with the first element
 100         either 'ADD_QUEUE', 'COMMAND', or 'KILL_QUEUE', the second element
 101         a UUID, and an optional third element of arbitrary type. The UUID
 102         identifies a receiver for replies.
 103
 104         An 'ADD_QUEUE' command should contain as third element a queue
 105         through which to send messages back to the sender of the
 106         command. A 'KILL_QUEUE' command removes the queue for that
 107         receiver from the list of queues through which to send replies.
 108
 109         A 'COMMAND' command is specified in greater detail by a string
 110         that is the tuple's third element. The game_command_handler takes
 111         care of processing this and sending out replies.
 112
 113         """
 114         while True:
 115             x = q.get()
 116             command_type = x[0]
 117             connection_id = x[1]
 118             content = None if len(x) == 2 else x[2]
 119             if command_type == 'ADD_QUEUE':
 120                 self.queues_out[connection_id] = content
 121             elif command_type == 'KILL_QUEUE':
 122                 del self.queues_out[connection_id]
 123             elif command_type == 'COMMAND':
 124                 self.handle_input(content, connection_id)
 125
 126     def run_loop_with_server(self):
 127         """Run connection of server talking to clients and game IO loop.
 128
 129         We have the TCP server (an instance of Server) and we have the
 130         game IO loop, a thread running self.loop. Both communicate with
 131         each other via a queue.Queue. While the TCP server may spawn
 132         parallel threads to many clients, the IO loop works sequentially
 133         through game commands received from the TCP server's threads (=
 134         client connections to the TCP server). A processed command may
 135         trigger messages to the commanding client or to all clients,
 136         delivered from the IO loop to the TCP server via the queue.
 137
 138         """
 139         q = queue.Queue()
 140         c = threading.Thread(target=self.loop, daemon=True, args=(q,))
 141         c.start()
 142         server = Server(q, 5000)
 143         try:
 144             server.serve_forever()
 145         except KeyboardInterrupt:
 146             pass
 147         finally:
 148             print('Killing server')
 149             server.server_close()
 150
 151     def handle_input(self, input_, connection_id=None, store=True):
 152         """Process input_ to command grammar, call command handler if found."""
 153         from inspect import signature
 154         import server_.game
 155
 156         def answer(connection_id, msg):
 157             if connection_id:
 158                 self.send(msg, connection_id)
 159             else:
 160                 print(msg)
 161
 162         try:
 163             command, args = self.parser.parse(input_)
 164             if command is None:
 165                 answer(connection_id, 'UNHANDLED_INPUT')
 166             else:
 167                 if 'connection_id' in list(signature(command).parameters):
 168                     command(*args, connection_id=connection_id)
 169                 else:
 170                     command(*args)
 171                     if store and not hasattr(command, 'dont_save'):
 172                         with open(self.game_file_name, 'a') as f:
 173                             f.write(input_ + '\n')
 174         except parser.ArgError as e:
 175             answer(connection_id, 'ARGUMENT_ERROR ' + quote(str(e)))
 176         except GameError as e:
 177             answer(connection_id, 'GAME_ERROR ' + quote(str(e)))
 178
 179     def send(self, msg, connection_id=None):
 180         """Send message msg to server's client(s) via self.queues_out.
 181
 182         If a specific client is identified by connection_id, only
 183         sends msg to that one. Else, sends it to all clients
 184         identified in self.queues_out.
 185
 186         """
 187         if connection_id:
 188             self.queues_out[connection_id].put(msg)
 189         else:
 190             for connection_id in self.queues_out:
 191                 self.queues_out[connection_id].put(msg)
 192
 193
 194 def quote(string):
 195     """Quote & escape string so client interprets it as single token."""
 196     quoted = []
 197     quoted += ['"']
 198     for c in string:
 199         if c in {'"', '\\'}:
 200             quoted += ['\\']
 201         quoted += [c]
 202     quoted += ['"']
 203     return ''.join(quoted)
 204
 205
 206 def stringify_yx(tuple_):
 207     """Transform tuple (y,x) into string 'Y:'+str(y)+',X:'+str(x)."""
 208     return 'Y:' + str(tuple_[0]) + ',X:' + str(tuple_[1])