- Servers here must have threaded behaviour. start() must not block,
- there is no run().
- """
- __is_started = False
- __servers = []
- __starter_threads = []
-
- # we don't want blocking server calls (think select()) to
- # wait forever and possibly prevent exiting the process,
- # but instead we want a form of polling/busy_wait pattern, where
- # _server_timeout should be used as the default timeout for
- # all I/O blocking operations
- _busywait_timeout = 0.5
-
- def __init__(self):
- Server.__servers.append(self)
- if Server.__is_started:
- # raise Exception('All instances of servers must be inited before the startAll()')
- # Since the startAll() won't be called again, allow this server to
- # init and then start it after 1sec (hopefully). Register that
- # timer thread in a list, so that we can abort the start if quitAll
- # is called in the meantime
- t = threading.Timer(1.0, self._late_start)
- t.name = 'Late start timer for %s' % str(self.__class__)
- Server.__starter_threads.append(t)
- t.start()
-
- def start(self):
- _logger.debug("called stub Server.start")
-
- def _late_start(self):
- self.start()
- for thr in Server.__starter_threads:
- if thr.finished.is_set():
- Server.__starter_threads.remove(thr)
-
- def stop(self):
- _logger.debug("called stub Server.stop")
-
- def stats(self):
- """ This function should return statistics about the server """
- return "%s: No statistics" % str(self.__class__)
-
- @classmethod
- def startAll(cls):
- if cls.__is_started:
- return
- _logger.info("Starting %d services" % len(cls.__servers))
- for srv in cls.__servers:
- srv.start()
- cls.__is_started = True
-
- @classmethod
- def quitAll(cls):
- if not cls.__is_started:
- return
- _logger.info("Stopping %d services" % len(cls.__servers))
- for thr in cls.__starter_threads:
- if not thr.finished.is_set():
- thr.cancel()
- cls.__starter_threads.remove(thr)
-
- for srv in cls.__servers:
- srv.stop()
- cls.__is_started = False
-
- @classmethod
- def allStats(cls):
- res = ["Servers %s" % ('stopped', 'started')[cls.__is_started]]
- res.extend(srv.stats() for srv in cls.__servers)
- return '\n'.join(res)
-
- def _close_socket(self):
- close_socket(self.socket)
-
-def replace_request_password(args):
- # password is always 3rd argument in a request, we replace it in RPC logs
- # so it's easier to forward logs for diagnostics/debugging purposes...
- if len(args) > 2:
- args = list(args)
- args[2] = '*'
- return tuple(args)
-
-def dispatch_rpc_log(logger, prefix, msg, depth=None):
- indent=''
- indent_after=' '*len(prefix)
- for line in (prefix+pformat(msg, depth=depth)).split('\n'):
- logger.debug(indent+line)
- indent=indent_after
-
-def dispatch_rpc(service_name, method, params):
- """ Handle a RPC call.
-
- This is pure Python code, the actual marshalling (from/to XML-RPC or
- NET-RPC) is done in a upper layer.
- """
- try:
- rpc_short = logging.getLogger(__name__ + '.rpc_short')
- rpc_full = logging.getLogger(__name__ + '.rpc_full')
- rpc_short_flag = rpc_short.isEnabledFor(logging.DEBUG)
- rpc_full_flag = rpc_full.isEnabledFor(logging.DEBUG)
- if rpc_short_flag or rpc_full_flag:
- start_time = time.time()
- if rpc_full_flag:
- dispatch_rpc_log(rpc_full,'%s.%s:request '%(service_name,method), replace_request_password(params))
-
- result = ExportService.getService(service_name).dispatch(method, params)
-
- if rpc_short_flag or rpc_full_flag:
- end_time = time.time()
- if rpc_full_flag:
- dispatch_rpc_log(rpc_full,'%s.%s:reply time:%.3fs '%(service_name,method,end_time - start_time), result)
- else:
- dispatch_rpc_log(rpc_short,'%s.%s time:%.3fs '%(service_name,method,end_time - start_time), replace_request_password(params), depth=1)