diff options
Diffstat (limited to 'synapse/push')
| -rw-r--r-- | synapse/push/__init__.py | 76 | ||||
| -rw-r--r-- | synapse/push/httppusher.py | 40 | ||||
| -rw-r--r-- | synapse/push/pusherpool.py | 94 |
3 files changed, 210 insertions, 0 deletions
diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py new file mode 100644 index 0000000000..df0b91a8e9 --- /dev/null +++ b/synapse/push/__init__.py @@ -0,0 +1,76 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from twisted.internet import defer + +from synapse.streams.config import PaginationConfig +from synapse.types import StreamToken + +import synapse.util.async + +import logging + +logger = logging.getLogger(__name__) + +class Pusher(object): + INITIAL_BACKOFF = 1000 + MAX_BACKOFF = 10 * 60 * 1000 + + def __init__(self, _hs, user_name, app, app_display_name, device_display_name, pushkey, data, last_token): + self.hs = _hs + self.evStreamHandler = self.hs.get_handlers().event_stream_handler + self.store = self.hs.get_datastore() + self.user_name = user_name + self.app = app + self.app_display_name = app_display_name + self.device_display_name = device_display_name + self.pushkey = pushkey + self.data = data + self.last_token = last_token + self.backoff_delay = Pusher.INITIAL_BACKOFF + + @defer.inlineCallbacks + def start(self): + if not self.last_token: + # First-time setup: get a token to start from (we can't just start from no token, ie. 'now' + # because we need the result to be reproduceable in case we fail to dispatch the push) + config = PaginationConfig(from_token=None, limit='1') + chunk = yield self.evStreamHandler.get_stream(self.user_name, config, timeout=0) + self.last_token = chunk['end'] + self.store.update_pusher_last_token(self.user_name, self.pushkey, self.last_token) + logger.info("Pusher %s for user %s starting from token %s", + self.pushkey, self.user_name, self.last_token) + + while True: + from_tok = StreamToken.from_string(self.last_token) + config = PaginationConfig(from_token=from_tok, limit='1') + chunk = yield self.evStreamHandler.get_stream(self.user_name, config, timeout=100*365*24*60*60*1000) + + if (self.dispatchPush(chunk['chunk'][0])): + self.backoff_delay = Pusher.INITIAL_BACKOFF + self.last_token = chunk['end'] + self.store.update_pusher_last_token(self.user_name, self.pushkey, self.last_token) + else: + logger.warn("Failed to dispatch push for user %s. Trying again in %dms", + self.user_name, self.backoff_delay) + yield synapse.util.async.sleep(self.backoff_delay / 1000.0) + self.backoff_delay *=2 + if self.backoff_delay > Pusher.MAX_BACKOFF: + self.backoff_delay = Pusher.MAX_BACKOFF + + +class PusherConfigException(Exception): + def __init__(self, msg): + super(PusherConfigException, self).__init__(msg) \ No newline at end of file diff --git a/synapse/push/httppusher.py b/synapse/push/httppusher.py new file mode 100644 index 0000000000..988c4e32f5 --- /dev/null +++ b/synapse/push/httppusher.py @@ -0,0 +1,40 @@ +# -*- coding: utf-8 -*- +# Copyright 2014 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from synapse.push import Pusher, PusherConfigException +from synapse.http.client import + +import logging + +logger = logging.getLogger(__name__) + +class HttpPusher(Pusher): + def __init__(self, _hs, user_name, app, app_display_name, device_display_name, pushkey, data, last_token): + super(HttpPusher, self).__init__(_hs, + user_name, + app, + app_display_name, + device_display_name, + pushkey, + data, + last_token) + if 'url' not in data: + raise PusherConfigException("'url' required in data for HTTP pusher") + self.url = data['url'] + + def dispatchPush(self, event): + print event + return True + diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py new file mode 100644 index 0000000000..436040f123 --- /dev/null +++ b/synapse/push/pusherpool.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +# Copyright 2014 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from twisted.internet import defer + +from httppusher import HttpPusher +from synapse.push import PusherConfigException + +import logging +import json + +logger = logging.getLogger(__name__) + +class PusherPool: + def __init__(self, _hs): + self.hs = _hs + self.store = self.hs.get_datastore() + self.pushers = [] + self.last_pusher_started = -1 + + def start(self): + self._pushers_added() + + def add_pusher(self, user_name, kind, app, 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, + # recreated, added and started: this means we have only one code path adding pushers. + self._create_pusher({ + "user_name": user_name, + "kind": kind, + "app": app, + "app_display_name": app_display_name, + "device_display_name": device_display_name, + "pushkey": pushkey, + "data": data, + "last_token": None + }) + self._add_pusher_to_store(user_name, kind, app, app_display_name, device_display_name, pushkey, data) + + @defer.inlineCallbacks + def _add_pusher_to_store(self, user_name, kind, app, app_display_name, device_display_name, pushkey, data): + yield self.store.add_pusher(user_name=user_name, + kind=kind, + app=app, + app_display_name=app_display_name, + device_display_name=device_display_name, + pushkey=pushkey, + data=json.dumps(data)) + self._pushers_added() + + def _create_pusher(self, pusherdict): + if pusherdict['kind'] == 'http': + return HttpPusher(self.hs, + user_name=pusherdict['user_name'], + app=pusherdict['app'], + app_display_name=pusherdict['app_display_name'], + device_display_name=pusherdict['device_display_name'], + pushkey=pusherdict['pushkey'], + data=pusherdict['data'], + last_token=pusherdict['last_token'] + ) + else: + raise PusherConfigException("Unknown pusher type '%s' for user %s" % + (pusherdict['kind'], pusherdict['user_name'])) + + @defer.inlineCallbacks + def _pushers_added(self): + pushers = yield self.store.get_all_pushers_after_id(self.last_pusher_started) + for p in pushers: + p['data'] = json.loads(p['data']) + if (len(pushers)): + self.last_pusher_started = pushers[-1]['id'] + + self._start_pushers(pushers) + + def _start_pushers(self, pushers): + logger.info("Starting %d pushers", (len(pushers))) + for pusherdict in pushers: + p = self._create_pusher(pusherdict) + if p: + self.pushers.append(p) + p.start() \ No newline at end of file |