1  import asyncio
   2  import logging
   3  import time
   4  import traceback
   5  
   6  from .compatibility import guarantee_single_callable
   7  
   8  logger = logging.getLogger(__name__)
   9  
  10  
  11  class StatelessServer:
  12      """
  13      Base server class that handles basic concepts like application instance
  14      creation/pooling, exception handling, and similar, for stateless protocols
  15      (i.e. ones without actual incoming connections to the process)
  16  
  17      Your code should override the handle() method, doing whatever it needs to,
  18      and calling get_or_create_application_instance with a unique `scope_id`
  19      and `scope` for the scope it wants to get.
  20  
  21      If an application instance is found with the same `scope_id`, you are
  22      given its input queue, otherwise one is made for you with the scope provided
  23      and you are given that fresh new input queue. Either way, you should do
  24      something like:
  25  
  26      input_queue = self.get_or_create_application_instance(
  27          "user-123456",
  28          {"type": "testprotocol", "user_id": "123456", "username": "andrew"},
  29      )
  30      input_queue.put_nowait(message)
  31  
  32      If you try and create an application instance and there are already
  33      `max_application` instances, the oldest/least recently used one will be
  34      reclaimed and shut down to make space.
  35  
  36      Application coroutines that error will be found periodically (every 100ms
  37      by default) and have their exceptions printed to the console. Override
  38      application_exception() if you want to do more when this happens.
  39  
  40      If you override run(), make sure you handle things like launching the
  41      application checker.
  42      """
  43  
  44      application_checker_interval = 0.1
  45  
  46      def __init__(self, application, max_applications=1000):
  47          # Parameters
  48          self.application = application
  49          self.max_applications = max_applications
  50          # Initialisation
  51          self.application_instances = {}
  52  
  53      ### Mainloop and handling
  54  
  55      def run(self):
  56          """
  57          Runs the asyncio event loop with our handler loop.
  58          """
  59          event_loop = asyncio.get_event_loop()
  60          asyncio.ensure_future(self.application_checker())
  61          try:
  62              event_loop.run_until_complete(self.handle())
  63          except KeyboardInterrupt:
  64              logger.info("Exiting due to Ctrl-C/interrupt")
  65  
  66      async def handle(self):
  67          raise NotImplementedError("You must implement handle()")
  68  
  69      async def application_send(self, scope, message):
  70          """
  71          Receives outbound sends from applications and handles them.
  72          """
  73          raise NotImplementedError("You must implement application_send()")
  74  
  75      ### Application instance management
  76  
  77      def get_or_create_application_instance(self, scope_id, scope):
  78          """
  79          Creates an application instance and returns its queue.
  80          """
  81          if scope_id in self.application_instances:
  82              self.application_instances[scope_id]["last_used"] = time.time()
  83              return self.application_instances[scope_id]["input_queue"]
  84          # See if we need to delete an old one
  85          while len(self.application_instances) > self.max_applications:
  86              self.delete_oldest_application_instance()
  87          # Make an instance of the application
  88          input_queue = asyncio.Queue()
  89          application_instance = guarantee_single_callable(self.application)
  90          # Run it, and stash the future for later checking
  91          future = asyncio.ensure_future(
  92              application_instance(
  93                  scope=scope,
  94                  receive=input_queue.get,
  95                  send=lambda message: self.application_send(scope, message),
  96              ),
  97          )
  98          self.application_instances[scope_id] = {
  99              "input_queue": input_queue,
 100              "future": future,
 101              "scope": scope,
 102              "last_used": time.time(),
 103          }
 104          return input_queue
 105  
 106      def delete_oldest_application_instance(self):
 107          """
 108          Finds and deletes the oldest application instance
 109          """
 110          oldest_time = min(
 111              details["last_used"] for details in self.application_instances.values()
 112          )
 113          for scope_id, details in self.application_instances.items():
 114              if details["last_used"] == oldest_time:
 115                  self.delete_application_instance(scope_id)
 116                  # Return to make sure we only delete one in case two have
 117                  # the same oldest time
 118                  return
 119  
 120      def delete_application_instance(self, scope_id):
 121          """
 122          Removes an application instance (makes sure its task is stopped,
 123          then removes it from the current set)
 124          """
 125          details = self.application_instances[scope_id]
 126          del self.application_instances[scope_id]
 127          if not details["future"].done():
 128              details["future"].cancel()
 129  
 130      async def application_checker(self):
 131          """
 132          Goes through the set of current application instance Futures and cleans up
 133          any that are done/prints exceptions for any that errored.
 134          """
 135          while True:
 136              await asyncio.sleep(self.application_checker_interval)
 137              for scope_id, details in list(self.application_instances.items()):
 138                  if details["future"].done():
 139                      exception = details["future"].exception()
 140                      if exception:
 141                          await self.application_exception(exception, details)
 142                      try:
 143                          del self.application_instances[scope_id]
 144                      except KeyError:
 145                          # Exception handling might have already got here before us. That's fine.
 146                          pass
 147  
 148      async def application_exception(self, exception, application_details):
 149          """
 150          Called whenever an application coroutine has an exception.
 151          """
 152          logging.error(
 153              "Exception inside application: %s\n%s%s",
 154              exception,
 155              "".join(traceback.format_tb(exception.__traceback__)),
 156              f"  {exception}",
 157          )
 158