public
Last active

Tornado demo application: TCP server to tic-tac-toe multiplayer game

  • Download Gist
tic-tac-toe.py
Python
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333
"""Simple TCP server for playing Tic-Tac-Toe game.
 
Use Player-to-Player game mode based on naming auth.
No thoughts about distribution or pub/sub mode
(for monitoring or something like this). Just
basic functionality.
"""
 
import time
import logging
import signal
import socket
 
from tornado import stack_context
from tornado.options import options, parse_command_line, define
from tornado.netutil import TCPServer
from tornado.ioloop import IOLoop
from tornado.util import b, bytes_type
 
 
class Game(object):
"""Single game representation"""
 
UNUSED_CELL = 0
 
class Winner(Exception):
"""Control exception for working with game end"""
def __init__(self, name):
self.name = name
super(Game.Winner, self).__init__('Winner!')
 
class Draw(Exception):
"""Control exception for working with game end"""
def __init__(self):
super(Game.Draw, self).__init__('Draw!')
 
# Current state of map
map = None
# List of possible steps of user
allowed_steps = dict([
('X', 1),
('O', -1)
])
def __init__(self, *players, **params):
self.players = dict(zip(self.allowed_steps.keys(), players))
self.params = params
logging.debug('Create game with params: %s', params)
self.first = 'X'
self._generate_map()
 
def step(self, sign, pos):
# Check allowed sign
try:
val = self.allowed_steps[sign]
except KeyError:
raise ValueError('Unknown step sign is given')
 
# Check map position
# We should make step with using 1-based numeration (not zero-based)
try:
if self.map[pos[0]-1][pos[1]-1] != Game.UNUSED_CELL:
raise ValueError('Position is already in use')
except IndexError:
raise ValueError('Illegal position for step')
 
# Make step
self.map[pos[0]-1][pos[1]-1] = val
 
# Recalculate game state:
# Check if some player is winner on no other step is possible
limit = self.params.get('win_limit', 3)
for i, row in enumerate(self.map):
if abs(sum(row)) == limit:
raise Game.Winner(self.players[sign].name)
 
if abs(sum([r[i] for r in self.map])) == limit:
raise Game.Winner(self.players[sign].name)
 
# Check all possible diagonals
up = len(self.map) + 1 - limit
for s in range(up):
if any([
(abs(sum([self.map[i+s][i+s] for i in range(limit)])) == limit),
(abs(sum([self.map[len(self.map) - (i+1+s)][i+s] for i in range(limit)])) == limit)
]):
raise Game.Winner(self.players[sign].name)
if sum([sum(map(abs, r)) for r in self.map]) == len(self.map)**2:
raise Game.Draw()
 
def render(self):
"""Return string of current state map representation"""
signs = dict([(v,s) for s,v in self.allowed_steps.items()])
return '\n'.join(map(
lambda r: ' '.join(map(lambda s: signs.get(s, '.'), r)),
self.map
))
 
def _generate_map(self):
assert self.map is None, "You couldn't regenerate map"
size = self.params.get('map_size', 3)
self.map = [[Game.UNUSED_CELL]*size for i in range(size)]
logging.debug('Generated map: %s', self.map)
 
class GameSessionMixin(object):
"""Handler of two-players interacion within game"""
 
waiter = None
 
def join(self):
"""Logic of joining to game session"""
# Check if there is at least one open game
# Connect to game if exist
if GameSessionMixin.waiter is not None:
enemy = GameSessionMixin.waiter
GameSessionMixin.run(enemy, self)
GameSessionMixin.waiter = None
else:
# Create new one and wait for new connections
GameSessionMixin.waiter = self
self.notify('Waiting for incoming player...')
 
@staticmethod
def run(*players):
"""Create game object and choose first runner"""
game = Game(*players,
map_size=options.map_size, win_limit=options.win_limit)
message = 'Game is starting. First step by %s.' % game.players[game.first]
for sign, p in game.players.iteritems():
p.game = game
p.sign = sign
if sign == game.first:
action = p.make_step
else:
action = p.wait_step
p.notify(message, callback=action)
 
def notify(self, note, callback=None):
"""Send game notifications"""
self.stream.write('%s, %s\n' % (self.name, note), callback=callback)
 
def make_step(self):
self.notify('make your step:', callback=self._on_make_step)
 
def _on_make_step(self):
"""Called when make step notification is send"""
self.stream.read_until(b("\n"), callback=self._on_step)
 
def _on_step(self, line):
"""Receive step from user"""
step = map(int, line.strip().split())
try:
self.game.step(self.sign, step)
except ValueError, e:
# Cycle for receiving normal step from user
self.notify(str(e), callback=self.make_step)
except Game.Winner, e:
GameSessionMixin.broadcast(self.game.players.values(),
'%s is WINNER!' % e.name,
'close')
except Game.Draw:
GameSessionMixin.broadcast(self.game.players.values(),
'DRAW in game!',
'close')
else:
battle = self.game.render() + '\n'
for sign, p in self.game.players.iteritems():
p.stream.write(battle,
callback=(p.wait_step if sign == self.sign else p.make_step))
 
def wait_step(self):
"""Just wait for step from other player.
 
Of course, we can notify user about this, but it will
complicate whole process, cause we will have problems
with handling async write operation.
"""
pass
def close(self):
"""Close players' stream"""
self.stream.write('Game over\n', callback=self.stream.close)
 
@staticmethod
def broadcast(sub, notification, callback_method):
"""Send notification to each player in ``sub`` list"""
for s in sub:
s.notify(notification.strip()+'\n',
callback=s.__getattribute__(callback_method))
 
 
class AuthMixin(object):
"""Batch of function for checking auth and registering users"""
 
# List of connected players
# TODO: Add periodic callback for checking player timeout
players = set()
 
def register(self, on_register=None):
"""Register player with using text name"""
if on_register:
self._register_callback = stack_context.wrap(on_register)
else:
self._register_callback = None
self.stream.write('Enter your name: ', callback=self._on_greeting)
 
def _on_greeting(self):
logging.debug('On greetings call')
self.stream.read_until(b("\n"), self._on_name)
 
def _on_name(self, line):
"""Ask user about name and save it in list of users"""
logging.debug('On name call with: %s', line)
name = line.strip()
if name not in self.__class__.players:
self.name = name
self.__class__.players.add(self)
if self._register_callback:
self._register_callback()
else:
message = 'Name %s is already in used. Choose another one:' % name
self.stream.write(message, callback=self._on_greeting)
 
def unregister(self):
"""Remove player from list of players"""
try:
self.__class__.players.remove(str(self))
except KeyError:
logging.warning('Try to remove illegal or undefined user')
 
class PlayerConnection(GameSessionMixin, AuthMixin):
"""Player logic handler"""
 
# Player's name. Should be setted for auth.
name = None
 
def __init__(self, stream, address, server):
"""Initialize base params and call stream reader for next line"""
self.stream = stream
if self.stream.socket.family not in (socket.AF_INET, socket.AF_INET6):
# Unix (or other) socket; fake the remote address
address = ('0.0.0.0', 0)
self.address = address
self.server = server
self.stream.set_close_callback(self._on_disconnect)
# Will block current stream flow until user's name is set
self.register(on_register=self.play)
 
def play(self):
"""Main logic function"""
self.join()
 
def _on_read(self, line):
"""Called when new line received from connection"""
# Some game logic (or magic)
self.wait()
 
def wait(self):
"""Read from stream until the next signed end of line"""
self.stream.read_until(b("\n"), self._on_read)
 
def _on_disconnect(self, *args, **kwargs):
"""Called on client disconnected"""
logging.info('Client disconnected %r', self.address)
# TODO: Should also check current game state and/or waiters status
self.unregister()
 
def __str__(self):
"""Build string representation, will be used for working with
server identity (not only name) in future"""
return str(self.name)
 
class TTTServer(TCPServer):
"""TCP server for handling incoming connections from players"""
 
def handle_stream(self, stream, address):
"""Called when new IOStream object is ready for usage"""
logging.info('Incoming connection from %r', address)
PlayerConnection(stream, address, server=self)
 
def sig_handler(sig, frame):
"""Catch signal and init callback.
More information about signal processing for graceful stoping
Tornado server you can find here:
http://codemehanika.org/blog/2011-10-28-graceful-stop-tornado.html
"""
logging.warning('Caught signal: %s', sig)
IOLoop.instance().add_callback(shutdown)
 
def shutdown():
"""Stop server and add callback to stop i/o loop"""
io_loop = IOLoop.instance()
 
logging.info('Stopping Tic-Tac-Toe tcp server')
io_loop.ttt.stop()
 
logging.info('Will shutdown in 2 seconds ...')
io_loop.add_timeout(time.time() + 2, io_loop.stop)
 
def main():
"""Main processing function"""
io_loop = IOLoop.instance()
 
# Create instance of Tic-Tac-Toe TCP server and save
# it as attribute of IOLoop instance. Of course, this
# is not the best way to spread ttt instance among
# several functions, but it's enough for demo app.
io_loop.ttt = TTTServer()
io_loop.ttt.listen(options.port)
 
# Init signals handler for TERM and INT signals
# (and so KeyboardInterrupt)
signal.signal(signal.SIGTERM, sig_handler)
signal.signal(signal.SIGINT, sig_handler)
 
logging.info('Starting TTT server on %d port', options.port)
io_loop.start()
 
define('debug', default=True, type=bool)
define('port', help='Port to listen to', default=8046, type=int)
define('map_size', help='Game map size', default=3, type=int)
define('win_limit', help='Signs to win', default=3, type=int)
 
if __name__ == '__main__':
parse_command_line()
main()

WTF?
val = self.__class__.allowed_steps[sign]
Using exceptions as flow control (Game.Draw, Game.Winner) -- bad. Never ever do that, future you will hate current you.
On line 59 you're using magic number 0, this is fragile.

Fixed using magic number.
Don't see any problem with using exceptions as flow control inside system based on 300 lines of code.

Please sign in to comment on this gist.

Something went wrong with that request. Please try again.