Skip to content

Instantly share code, notes, and snippets.

@veox
Last active Jan 19, 2019
Embed
What would you like to do?
#!/usr/bin/env python
import logging
from pprint import pprint as pp
from typing import (
Dict,
List,
Tuple,
Type,
)
class Command:
_cmd_id = None
_cmd_id_offset = None
_snappy_support = None
@property
def cmd_id(cls):
return cls._cmd_id_offset + cls._cmd_id
class Hello(Command):
_cmd_id = 0
class Disconnect(Command):
_cmd_id = 1
class Ping(Command):
_cmd_id = 2
class Pong(Command):
_cmd_id = 3
command_classes: Dict[Tuple, Type[Command]] = {}
def init_command_class(cmd_class, cmd_id_offset, snappy_support):
id = (cmd_class, cmd_id_offset, snappy_support)
print('XXXXX', id)
# use existing if available
if id in command_classes.keys():
print('XXXXX IN COMMAND CLASS CACHE')
return command_classes[id]
class _cmd_class(cmd_class):
_cmd_id_offset = cmd_id_offset
_snappy_support = snappy_support
def __repr__(self):
return f'{id}'
# def __type__(self):
# return cmd_class
c = _cmd_class()
command_classes[id] = c
return c
class Protocol:
name: str = None
version: int = None
_commands: List[Type[Command]] = []
_logger: logging.Logger = None
def __init__(self, cmd_id_offset: int, snappy_support: bool) -> None:
self._cmd_id_offset = cmd_id_offset
self._snappy_support = snappy_support
self._update_protocol_commands()
return
@property
def logger(self) -> logging.Logger:
if self._logger is None:
self._logger = logging.getLogger(f'{type(self).__name__}')
return self._logger
def _update_protocol_commands(self) -> None:
self.logger.debug('updating commands for %r', self)
self.commands = [init_command_class(cmd_class, self._cmd_id_offset, self._snappy_support)
for cmd_class in self._commands]
pp('commands:')
pp(self.commands)
self.cmd_by_type = {type(cmd): cmd for cmd in self.commands}
pp('cmd_by_type:')
pp(self.cmd_by_type)
self.cmd_by_id = {cmd.cmd_id: cmd for cmd in self.commands}
pp('cmd_by_id:')
pp(self.cmd_by_id)
return
def supports_command(self, cmd_type: Type[Command]) -> bool:
return cmd_type in self._commands
class P2PProtocol(Protocol):
name = 'p2p'
version = 5
_commands = Protocol._commands + [Hello, Ping, Pong, Disconnect]
def __init__(self, cmd_id_offset: int = 0, snappy_support: bool = False) -> None:
super().__init__(cmd_id_offset, snappy_support)
return
if __name__ == '__main__':
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('main')
p0 = Protocol(0, False)
protocols = [
P2PProtocol(16, False),
P2PProtocol(16, False),
P2PProtocol(16, True),
]
logger.info('='*72)
logger.info('base class instance: %r', p0)
logger.info('p2p class instances: %r', protocols)
logger.info('command classes: %r', command_classes)
logger.info('command count across all protocol instances: %s', len(command_classes))
for protocol in [p0] + protocols:
logger.info('%s supports Hello: %s', protocol, protocol.supports_command(Hello))
for command in protocol.commands:
logger.info('%r %s %s', command, command.cmd_id, command._snappy_support)
DEBUG:Protocol:updating commands for <__main__.Protocol object at 0x7f0647032dd8>
'commands:'
[]
'cmd_by_type:'
{}
'cmd_by_id:'
{}
DEBUG:P2PProtocol:updating commands for <__main__.P2PProtocol object at 0x7f0647032e48>
XXXXX (<class '__main__.Hello'>, 16, False)
XXXXX (<class '__main__.Ping'>, 16, False)
XXXXX (<class '__main__.Pong'>, 16, False)
XXXXX (<class '__main__.Disconnect'>, 16, False)
'commands:'
[(<class '__main__.Hello'>, 16, False),
(<class '__main__.Ping'>, 16, False),
(<class '__main__.Pong'>, 16, False),
(<class '__main__.Disconnect'>, 16, False)]
'cmd_by_type:'
{<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Hello'>, 16, False),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Ping'>, 16, False),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Pong'>, 16, False),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Disconnect'>, 16, False)}
'cmd_by_id:'
{16: (<class '__main__.Hello'>, 16, False),
17: (<class '__main__.Disconnect'>, 16, False),
18: (<class '__main__.Ping'>, 16, False),
19: (<class '__main__.Pong'>, 16, False)}
DEBUG:P2PProtocol:updating commands for <__main__.P2PProtocol object at 0x7f0647032e80>
XXXXX (<class '__main__.Hello'>, 16, False)
XXXXX IN COMMAND CLASS CACHE
XXXXX (<class '__main__.Ping'>, 16, False)
XXXXX IN COMMAND CLASS CACHE
XXXXX (<class '__main__.Pong'>, 16, False)
XXXXX IN COMMAND CLASS CACHE
XXXXX (<class '__main__.Disconnect'>, 16, False)
XXXXX IN COMMAND CLASS CACHE
'commands:'
[(<class '__main__.Hello'>, 16, False),
(<class '__main__.Ping'>, 16, False),
(<class '__main__.Pong'>, 16, False),
(<class '__main__.Disconnect'>, 16, False)]
'cmd_by_type:'
{<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Hello'>, 16, False),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Ping'>, 16, False),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Pong'>, 16, False),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Disconnect'>, 16, False)}
'cmd_by_id:'
{16: (<class '__main__.Hello'>, 16, False),
17: (<class '__main__.Disconnect'>, 16, False),
18: (<class '__main__.Ping'>, 16, False),
19: (<class '__main__.Pong'>, 16, False)}
DEBUG:P2PProtocol:updating commands for <__main__.P2PProtocol object at 0x7f0647032fd0>
XXXXX (<class '__main__.Hello'>, 16, True)
XXXXX (<class '__main__.Ping'>, 16, True)
XXXXX (<class '__main__.Pong'>, 16, True)
XXXXX (<class '__main__.Disconnect'>, 16, True)
'commands:'
[(<class '__main__.Hello'>, 16, True),
(<class '__main__.Ping'>, 16, True),
(<class '__main__.Pong'>, 16, True),
(<class '__main__.Disconnect'>, 16, True)]
'cmd_by_type:'
{<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Hello'>, 16, True),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Ping'>, 16, True),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Pong'>, 16, True),
<class '__main__.init_command_class.<locals>._cmd_class'>: (<class '__main__.Disconnect'>, 16, True)}
'cmd_by_id:'
{16: (<class '__main__.Hello'>, 16, True),
17: (<class '__main__.Disconnect'>, 16, True),
18: (<class '__main__.Ping'>, 16, True),
19: (<class '__main__.Pong'>, 16, True)}
INFO:main:========================================================================
INFO:main:base class instance: <__main__.Protocol object at 0x7f0647032dd8>
INFO:main:p2p class instances: [<__main__.P2PProtocol object at 0x7f0647032e48>, <__main__.P2PProtocol object at 0x7f0647032e80>, <__main__.P2PProtocol object at 0x7f0647032fd0>]
INFO:main:command classes: {(<class '__main__.Hello'>, 16, False): (<class '__main__.Hello'>, 16, False), (<class '__main__.Ping'>, 16, False): (<class '__main__.Ping'>, 16, False), (<class '__main__.Pong'>, 16, False): (<class '__main__.Pong'>, 16, False), (<class '__main__.Disconnect'>, 16, False): (<class '__main__.Disconnect'>, 16, False), (<class '__main__.Hello'>, 16, True): (<class '__main__.Hello'>, 16, True), (<class '__main__.Ping'>, 16, True): (<class '__main__.Ping'>, 16, True), (<class '__main__.Pong'>, 16, True): (<class '__main__.Pong'>, 16, True), (<class '__main__.Disconnect'>, 16, True): (<class '__main__.Disconnect'>, 16, True)}
INFO:main:command count across all protocol instances: 8
INFO:main:<__main__.Protocol object at 0x7f0647032dd8> supports Hello: False
INFO:main:<__main__.P2PProtocol object at 0x7f0647032e48> supports Hello: True
INFO:main:(<class '__main__.Hello'>, 16, False) 16 False
INFO:main:(<class '__main__.Ping'>, 16, False) 18 False
INFO:main:(<class '__main__.Pong'>, 16, False) 19 False
INFO:main:(<class '__main__.Disconnect'>, 16, False) 17 False
INFO:main:<__main__.P2PProtocol object at 0x7f0647032e80> supports Hello: True
INFO:main:(<class '__main__.Hello'>, 16, False) 16 False
INFO:main:(<class '__main__.Ping'>, 16, False) 18 False
INFO:main:(<class '__main__.Pong'>, 16, False) 19 False
INFO:main:(<class '__main__.Disconnect'>, 16, False) 17 False
INFO:main:<__main__.P2PProtocol object at 0x7f0647032fd0> supports Hello: True
INFO:main:(<class '__main__.Hello'>, 16, True) 16 True
INFO:main:(<class '__main__.Ping'>, 16, True) 18 True
INFO:main:(<class '__main__.Pong'>, 16, True) 19 True
INFO:main:(<class '__main__.Disconnect'>, 16, True) 17 True
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment