Source code for pyspades.protocol

# Copyright (c) Mathias Kaerlev 2011-2012.

# This file is part of pyspades.

# pyspades is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# pyspades is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with pyspades.  If not, see <>.

import asyncio
from twisted.internet import reactor
from pyspades.bytes import ByteWriter

import enet

[docs]class BaseConnection: disconnected = False timeout_call = None def __init__(self, protocol, peer): self.protocol = protocol self.peer = peer
[docs] def timed_out(self): self.disconnect()
[docs] def disconnect(self, data=0): if self.disconnected: return self.disconnected = True self.peer.disconnect(data) self.protocol.remove_peer(self.peer) self.on_disconnect()
[docs] def loader_received(self, loader): raise NotImplementedError('loader_received() not implemented')
[docs] def send_contained(self, contained, sequence=False): if self.disconnected: return if sequence: flags = enet.PACKET_FLAG_UNSEQUENCED else: flags = enet.PACKET_FLAG_RELIABLE data = ByteWriter() contained.write(data) packet = enet.Packet(bytes(data), flags) self.peer.send(0, packet)
# events
[docs] def on_connect(self): pass
[docs] def on_disconnect(self): pass
# properties @property def latency(self): return self.peer.roundTripTime
[docs]class BaseProtocol: connection_class = BaseConnection max_connections = 33 is_client = False def __init__(self, port=None, interface=b'*', update_interval=1 / 60.0): if port is not None and interface is not None: address = enet.Address(interface, port) else: address = None try: = enet.Host(address, self.max_connections, 1) except MemoryError: # pyenet raises memoryerror when the enet host could not be created raise IOError("Failed to Create Enet Host. Is the Port in use?") self.update_loop = asyncio.ensure_future(self.update()) self.connections = {} self.clients = {}
[docs] def connect(self, connection_class, host, port, version, channel_count=1, timeout=5.0): host = host.encode() peer =, port), channel_count, version) connection = connection_class(self, peer) connection.timeout_call = reactor.callLater(timeout, connection.timed_out) self.clients[peer] = connection return connection
[docs] def on_connect(self, peer): connection = self.connection_class(self, peer) self.connections[peer] = connection connection.on_connect()
[docs] def on_disconnect(self, peer): try: connection = self.connections.pop(peer) connection.disconnected = True connection.on_disconnect() except KeyError: return
[docs] def data_received(self, peer, packet): connection = self.connections[peer] connection.loader_received(packet)
[docs] def remove_peer(self, peer): if peer in self.connections: del self.connections[peer] elif peer in self.clients: del self.clients[peer] self.check_client()
[docs] def check_client(self): if self.is_client and not self.clients: self.update_loop.stop() self.update_loop = None = None # important for GC
[docs] def update(self): try: while 1: if is None: return try: event = except IOError: break if event is None: break event_type = event.type if event_type == enet.EVENT_TYPE_NONE: break peer = event.peer is_client = peer in self.clients if is_client: connection = self.clients[peer] if event_type == enet.EVENT_TYPE_CONNECT: connection.on_connect() connection.timeout_call.cancel() elif event_type == enet.EVENT_TYPE_DISCONNECT: connection.on_disconnect() del self.clients[peer] self.check_client() elif event.type == enet.EVENT_TYPE_RECEIVE: connection.loader_received(event.packet) else: if event_type == enet.EVENT_TYPE_CONNECT: self.on_connect(peer) elif event_type == enet.EVENT_TYPE_DISCONNECT: self.on_disconnect(peer) elif event.type == enet.EVENT_TYPE_RECEIVE: self.data_received(peer, event.packet) except: # make sure the LoopingCall doesn't catch this and stops import traceback traceback.print_exc()