diff options
Diffstat (limited to 'synapse/push')
| -rw-r--r-- | synapse/push/__init__.py | 12 | ||||
| -rw-r--r-- | synapse/push/httppusher.py | 25 | ||||
| -rw-r--r-- | synapse/push/pusherpool.py | 47 |
3 files changed, 48 insertions, 36 deletions
diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py index 5fca3bd772..5fe8719fe7 100644 --- a/synapse/push/__init__.py +++ b/synapse/push/__init__.py @@ -30,7 +30,7 @@ class Pusher(object): MAX_BACKOFF = 60 * 60 * 1000 GIVE_UP_AFTER = 24 * 60 * 60 * 1000 - def __init__(self, _hs, user_name, app_id, app_instance_id, + def __init__(self, _hs, user_name, app_id, app_display_name, device_display_name, pushkey, data, last_token, last_success, failing_since): self.hs = _hs @@ -39,7 +39,6 @@ class Pusher(object): self.clock = self.hs.get_clock() self.user_name = user_name self.app_id = app_id - self.app_instance_id = app_instance_id self.app_display_name = app_display_name self.device_display_name = device_display_name self.pushkey = pushkey @@ -48,6 +47,7 @@ class Pusher(object): self.last_success = last_success # not actually used self.backoff_delay = Pusher.INITIAL_BACKOFF self.failing_since = failing_since + self.alive = True @defer.inlineCallbacks def start(self): @@ -65,7 +65,7 @@ class Pusher(object): logger.info("Pusher %s for user %s starting from token %s", self.pushkey, self.user_name, self.last_token) - while True: + while self.alive: from_tok = StreamToken.from_string(self.last_token) config = PaginationConfig(from_token=from_tok, limit='1') chunk = yield self.evStreamHandler.get_stream( @@ -81,6 +81,9 @@ class Pusher(object): if not single_event: continue + if not self.alive: + continue + ret = yield self.dispatch_push(single_event) if ret: self.backoff_delay = Pusher.INITIAL_BACKOFF @@ -142,6 +145,9 @@ class Pusher(object): if self.backoff_delay > Pusher.MAX_BACKOFF: self.backoff_delay = Pusher.MAX_BACKOFF + def stop(self): + self.alive = False + def dispatch_push(self, p): pass diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py index fd7fe4e39c..f94f673391 100644 --- a/synapse/push/httppusher.py +++ b/synapse/push/httppusher.py @@ -24,14 +24,13 @@ logger = logging.getLogger(__name__) class HttpPusher(Pusher): - def __init__(self, _hs, user_name, app_id, app_instance_id, + def __init__(self, _hs, user_name, app_id, app_display_name, device_display_name, pushkey, data, last_token, last_success, failing_since): super(HttpPusher, self).__init__( _hs, user_name, app_id, - app_instance_id, app_display_name, device_display_name, pushkey, @@ -69,16 +68,18 @@ class HttpPusher(Pusher): # we may have to fetch this over federation and we # can't trust it anyway: is it worth it? #'fromDisplayName': 'Steve Stevington' - }, - #'counts': { -- we don't mark messages as read yet so - # we have no way of knowing - # 'unread': 1, - # 'missedCalls': 2 - # }, - 'devices': { - self.pushkey: { - 'data': self.data_minus_url - } + #'counts': { -- we don't mark messages as read yet so + # we have no way of knowing + # 'unread': 1, + # 'missedCalls': 2 + # }, + 'devices': [ + { + 'app_id': self.app_id, + 'pushkey': self.pushkey, + 'data': self.data_minus_url + } + ] } } diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py index 045c36f3b7..d34ef3f6cf 100644 --- a/synapse/push/pusherpool.py +++ b/synapse/push/pusherpool.py @@ -24,17 +24,23 @@ import json logger = logging.getLogger(__name__) + class PusherPool: def __init__(self, _hs): self.hs = _hs self.store = self.hs.get_datastore() - self.pushers = [] + self.pushers = {} self.last_pusher_started = -1 + @defer.inlineCallbacks def start(self): - self._pushers_added() + pushers = yield self.store.get_all_pushers() + for p in pushers: + p['data'] = json.loads(p['data']) + self._start_pushers(pushers) - def add_pusher(self, user_name, kind, app_id, app_instance_id, + @defer.inlineCallbacks + def add_pusher(self, user_name, kind, app_id, app_display_name, device_display_name, pushkey, data): # we try to create the pusher just to validate the config: it # will then get pulled out of the database, @@ -44,7 +50,6 @@ class PusherPool: "user_name": user_name, "kind": kind, "app_id": app_id, - "app_instance_id": app_instance_id, "app_display_name": app_display_name, "device_display_name": device_display_name, "pushkey": pushkey, @@ -53,25 +58,26 @@ class PusherPool: "last_success": None, "failing_since": None }) - self._add_pusher_to_store(user_name, kind, app_id, app_instance_id, - app_display_name, device_display_name, - pushkey, data) + yield self._add_pusher_to_store( + user_name, kind, app_id, + app_display_name, device_display_name, + pushkey, data + ) @defer.inlineCallbacks - def _add_pusher_to_store(self, user_name, kind, app_id, app_instance_id, + def _add_pusher_to_store(self, user_name, kind, app_id, app_display_name, device_display_name, pushkey, data): yield self.store.add_pusher( user_name=user_name, kind=kind, app_id=app_id, - app_instance_id=app_instance_id, app_display_name=app_display_name, device_display_name=device_display_name, pushkey=pushkey, data=json.dumps(data) ) - self._pushers_added() + self._refresh_pusher((app_id, pushkey)) def _create_pusher(self, pusherdict): if pusherdict['kind'] == 'http': @@ -79,7 +85,6 @@ class PusherPool: self.hs, user_name=pusherdict['user_name'], app_id=pusherdict['app_id'], - app_instance_id=pusherdict['app_instance_id'], app_display_name=pusherdict['app_display_name'], device_display_name=pusherdict['device_display_name'], pushkey=pusherdict['pushkey'], @@ -95,21 +100,21 @@ class PusherPool: ) @defer.inlineCallbacks - def _pushers_added(self): - pushers = yield self.store.get_all_pushers_after_id( - self.last_pusher_started + def _refresh_pusher(self, app_id_pushkey): + p = yield self.store.get_pushers_by_app_id_and_pushkey( + app_id_pushkey ) - for p in pushers: - p['data'] = json.loads(p['data']) - if len(pushers): - self.last_pusher_started = pushers[-1]['id'] + p['data'] = json.loads(p['data']) - self._start_pushers(pushers) + self._start_pushers([p]) def _start_pushers(self, pushers): - logger.info("Starting %d pushers", (len(pushers))) + logger.info("Starting %d pushers", len(pushers)) for pusherdict in pushers: p = self._create_pusher(pusherdict) if p: - self.pushers.append(p) + fullid = "%s:%s" % (pusherdict['app_id'], pusherdict['pushkey']) + if fullid in self.pushers: + self.pushers[fullid].stop() + self.pushers[fullid] = p p.start() |