import errno
from . import log
[docs]class StaticCache:
SERVER_CLASS = None
SERVER_KWDS = {}
CACHE = None
[docs] @classmethod
def cache(cls):
if not cls.CACHE:
cls.CACHE = ServerCache(cls.SERVER_CLASS, **cls.SERVER_KWDS)
return cls.CACHE
[docs] @classmethod
def close_all(cls):
if cls.CACHE:
cls.CACHE.close_all()
cls.CACHE = None
[docs]class ServerCache:
"""
A class that caches servers by key so you don't keep closing and re-opening
the same server and interrupting your connection.
The exact nature of the key depends on the sort of server.
For example, for a server socket like SimPixel, it would be just a port
number, whereas for a UDP connection like Art-Net, it would be a
port, ip_address pair.
"""
def __init__(self, constructor, **kwds):
"""
:param constructor: a function which takes a key and some keywords,
and returns a new server
:param kwds: keywords to the ``constructor`` function
"""
self.servers = {}
self.constructor = constructor
self.kwds = kwds
[docs] def get_server(self, key, **kwds):
"""
Get a new or existing server for this key.
:param int key: key for the server to use
"""
kwds = dict(self.kwds, **kwds)
server = self.servers.get(key)
if server:
# Make sure it's the right server.
server.check_keywords(self.constructor, kwds)
else:
# Make a new server
server = _CachedServer(self.constructor, key, kwds)
self.servers[key] = server
return server
[docs] def close(self, key):
server = self.servers.pop(key, None)
if server:
server.server.close()
return True
[docs] def close_all(self):
for key in list(self.servers.keys()):
self.close(key)
class _CachedServer:
def __init__(self, constructor, key, kwds):
try:
self.server = constructor(key, **kwds)
except OSError as e:
if e.errno == errno.EADDRINUSE:
e.strerror += ADDRESS_IN_USE_ERROR.format(key)
e.args = (e.errno, e.strerror)
raise
self.key = key
self.constructor = constructor
self.kwds = kwds
def check_keywords(self, constructor, kwds):
if self.constructor != constructor:
raise ValueError(CACHED_SERVER_ERROR.format(
key=self.key,
new_type=str(constructor),
old_type=str(self.constructor)))
if self.kwds != kwds:
log.warning(CACHED_KWDS_WARNING.format(server=self, kwds=kwds))
def close(self):
pass
def __getattr__(self, key):
# Pass through all other attributes to the server.
return getattr(self.server, key)
ADDRESS_IN_USE_ERROR = """
Cached server {0} on your machine is already in use.
Perhaps BiblioPixel is already running on your machine?
"""
CACHED_SERVER_ERROR = """
Tried to open server of type {new_type} on {port}, but there was already
a server of type {old_type} running there.
"""
CACHED_KWDS_WARNING = """
Cached server for {server.port} had keywords {server.kwds},
but keywords {kwds} were requested.
"""