docs: async def observer_evaluate(self, message):
          """Execute observer evaluation on the worker or throttle."""
          observer_id = message['observer']
          throttle_rate = get_queryobserver_settings()['throttle_rate']
          if throttle_rate <= 0:
              await self._evaluate(observer_id)
              return
  
          cache_key = throttle_cache_key(observer_id)
          try:
              count = cache.incr(cache_key)
              # Ignore if delayed observer already scheduled.
              if count == 2:
                  await self.channel_layer.send(
                      CHANNEL_MAIN,
                      {
                          'type': TYPE_POLL,
                          'observer': observer_id,
                          'interval': throttle_rate,
                      },
                  )
          except ValueError:
              count = cache.get_or_set(cache_key, default=1, timeout=throttle_rate)
              # Ignore if cache was set and increased in another thread.
              if count == 1:
                  await self._evaluate(observer_id)