3 from __future__ import absolute_import, print_function
10 from .config import actor_class
12 def _notify_subscribers(response, subscribers):
13 """Send the response to all the subscriber methods.
15 If any of the subscriber actors have stopped, remove them from the
18 dead_subscribers = set()
19 for subscriber in subscribers:
22 except pykka.ActorDeadError:
23 dead_subscribers.add(subscriber)
24 subscribers.difference_update(dead_subscribers)
26 class RemotePollLoopActor(actor_class):
27 """Abstract actor class to regularly poll a remote service.
29 This actor sends regular requests to a remote service, and sends each
30 response to subscribers. It takes care of error handling, and retrying
31 requests with exponential backoff.
33 To use this actor, define the _send_request method. If you also
34 define an _item_key method, this class will support subscribing to
35 a specific item by key in responses.
37 def __init__(self, client, timer_actor, poll_wait=60, max_poll_wait=180):
38 super(RemotePollLoopActor, self).__init__()
40 self._timer = timer_actor
41 self._later = self.actor_ref.proxy()
42 self._polling_started = False
43 self.min_poll_wait = poll_wait
44 self.max_poll_wait = max_poll_wait
45 self.poll_wait = self.min_poll_wait
46 self.all_subscribers = set()
47 self.key_subscribers = {}
48 if hasattr(self, '_item_key'):
49 self.subscribe_to = self._subscribe_to
52 self._logger = logging.getLogger("%s.%s" % (self.__class__.__name__, id(self.actor_urn[9:])))
54 def _start_polling(self):
55 if not self._polling_started:
56 self._polling_started = True
59 def subscribe(self, subscriber):
60 self.all_subscribers.add(subscriber)
61 self._logger.debug("%s subscribed to all events", subscriber.actor_ref.actor_urn)
64 # __init__ exposes this method to the proxy if the subclass defines
66 def _subscribe_to(self, key, subscriber):
67 self.key_subscribers.setdefault(key, set()).add(subscriber)
68 self._logger.debug("%s subscribed to events for '%s'", subscriber.actor_ref.actor_urn, key)
71 def _send_request(self):
72 raise NotImplementedError("subclasses must implement request method")
74 def _got_response(self, response):
75 self._logger.debug("got response with %d items", len(response))
76 self.poll_wait = self.min_poll_wait
77 _notify_subscribers(response, self.all_subscribers)
78 if hasattr(self, '_item_key'):
79 items = {self._item_key(x): x for x in response}
80 for key, subscribers in self.key_subscribers.iteritems():
81 _notify_subscribers(items.get(key), subscribers)
83 def _got_error(self, error):
84 self.poll_wait = min(self.poll_wait * 2, self.max_poll_wait)
85 return "got error: {} - will try again in {} seconds".format(
86 error, self.poll_wait)
88 def is_common_error(self, exception):
91 def poll(self, scheduled_start=None):
92 self._logger.debug("sending request")
93 start_time = time.time()
94 if scheduled_start is None:
95 scheduled_start = start_time
97 response = self._send_request()
98 except Exception as error:
99 errmsg = self._got_error(error)
100 if self.is_common_error(error):
101 self._logger.warning(errmsg)
103 self._logger.exception(errmsg)
104 next_poll = start_time + self.poll_wait
106 self._got_response(response)
107 next_poll = scheduled_start + self.poll_wait
108 self._logger.info("request took %s seconds", (time.time() - scheduled_start))
109 end_time = time.time()
110 if next_poll < end_time: # We've drifted too much; start fresh.
111 next_poll = end_time + self.poll_wait
112 self._timer.schedule(next_poll, self._later.poll, next_poll)