diff --git a/synapse/api/errors.py b/synapse/api/errors.py
index 581439ceb3..64784bf212 100644
--- a/synapse/api/errors.py
+++ b/synapse/api/errors.py
@@ -34,6 +34,7 @@ class Codes(object):
LIMIT_EXCEEDED = "M_LIMIT_EXCEEDED"
CAPTCHA_NEEDED = "M_CAPTCHA_NEEDED"
CAPTCHA_INVALID = "M_CAPTCHA_INVALID"
+ MISSING_PARAM = "M_MISSING_PARAM"
class CodeMessageException(Exception):
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index 133b4521ba..a5833b2910 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -26,7 +26,7 @@ from twisted.web.server import Site
from synapse.http.server import JsonResource, RootRedirect
from synapse.http.content_repository import ContentRepoResource
from synapse.http.server_key_resource import LocalKey
-from synapse.http.client import MatrixHttpClient
+from synapse.http.client import MatrixFederationHttpClient
from synapse.api.urls import (
CLIENT_PREFIX, FEDERATION_PREFIX, WEB_CLIENT_PREFIX, CONTENT_REPO_PREFIX,
SERVER_KEY_PREFIX,
@@ -51,7 +51,7 @@ logger = logging.getLogger(__name__)
class SynapseHomeServer(HomeServer):
def build_http_client(self):
- return MatrixHttpClient(self)
+ return MatrixFederationHttpClient(self)
def build_resource_for_client(self):
return JsonResource()
@@ -242,6 +242,8 @@ def setup():
bind_port = None
hs.start_listening(bind_port, config.unsecure_port)
+ hs.get_pusherpool().start()
+
if config.daemonize:
print config.pid_file
daemon = Daemonize(
diff --git a/synapse/handlers/login.py b/synapse/handlers/login.py
index 99d15261d4..1204dc3b8f 100644
--- a/synapse/handlers/login.py
+++ b/synapse/handlers/login.py
@@ -17,7 +17,7 @@ from twisted.internet import defer
from ._base import BaseHandler
from synapse.api.errors import LoginError, Codes
-from synapse.http.client import IdentityServerHttpClient
+from synapse.http.client import SimpleHttpClient
from synapse.util.emailutils import EmailException
import synapse.util.emailutils as emailutils
@@ -97,7 +97,7 @@ class LoginHandler(BaseHandler):
@defer.inlineCallbacks
def _query_email(self, email):
- httpCli = IdentityServerHttpClient(self.hs)
+ httpCli = SimpleHttpClient(self.hs)
data = yield httpCli.get_json(
'matrix.org:8090', # TODO FIXME This should be configurable.
"/_matrix/identity/api/v1/lookup?medium=email&address=" +
diff --git a/synapse/handlers/register.py b/synapse/handlers/register.py
index c59ac1a3c8..122bf065c9 100644
--- a/synapse/handlers/register.py
+++ b/synapse/handlers/register.py
@@ -22,7 +22,7 @@ from synapse.api.errors import (
)
from ._base import BaseHandler
import synapse.util.stringutils as stringutils
-from synapse.http.client import IdentityServerHttpClient
+from synapse.http.client import SimpleHttpClient
from synapse.http.client import CaptchaServerHttpClient
import base64
@@ -159,7 +159,7 @@ class RegistrationHandler(BaseHandler):
def _threepid_from_creds(self, creds):
# TODO: get this from the homeserver rather than creating a new one for
# each request
- httpCli = IdentityServerHttpClient(self.hs)
+ httpCli = SimpleHttpClient(self.hs)
# XXX: make this configurable!
trustedIdServers = ['matrix.org:8090']
if not creds['idServer'] in trustedIdServers:
@@ -178,7 +178,7 @@ class RegistrationHandler(BaseHandler):
@defer.inlineCallbacks
def _bind_threepid(self, creds, mxid):
- httpCli = IdentityServerHttpClient(self.hs)
+ httpCli = SimpleHttpClient(self.hs)
data = yield httpCli.post_urlencoded_get_json(
creds['idServer'],
"/_matrix/identity/api/v1/3pid/bind",
diff --git a/synapse/http/client.py b/synapse/http/client.py
index dea61ba1e0..6361ac55f9 100644
--- a/synapse/http/client.py
+++ b/synapse/http/client.py
@@ -154,16 +154,81 @@ class BaseHttpClient(object):
defer.returnValue(response)
-class MatrixHttpClient(BaseHttpClient):
- """ Wrapper around the twisted HTTP client api. Implements
+class SimpleHttpClient(BaseHttpClient):
+ """
+ A simple, no-frills HTTP client with methods that wrap up common ways of using HTTP in Matrix
+ """
+ def _getEndpoint(self, reactor, destination):
+ return matrix_endpoint(reactor, destination, timeout=10)
+
+ @defer.inlineCallbacks
+ def post_urlencoded_get_json(self, destination, path, args={}):
+ logger.debug("post_urlencoded_get_json args: %s", args)
+ query_bytes = urllib.urlencode(args, True)
+
+ def body_callback(method, url_bytes, headers_dict):
+ return FileBodyProducer(StringIO(query_bytes))
+
+ response = yield self._create_request(
+ destination.encode("ascii"),
+ "POST",
+ path.encode("ascii"),
+ body_callback=body_callback,
+ headers_dict={
+ "Content-Type": ["application/x-www-form-urlencoded"]
+ }
+ )
+
+ body = yield readBody(response)
+
+ defer.returnValue(json.loads(body))
+
+ @defer.inlineCallbacks
+ def get_json(self, destination, path, args={}, retry_on_dns_fail=True):
+ """ Get's some json from the given host and path
+
+ Args:
+ destination (str): The remote server to send the HTTP request to.
+ path (str): The HTTP path.
+ args (dict): A dictionary used to create query strings, defaults to
+ None.
+ **Note**: The value of each key is assumed to be an iterable
+ and *not* a string.
+
+ Returns:
+ Deferred: Succeeds when we get *any* HTTP response.
+
+ The result of the deferred is a tuple of `(code, response)`,
+ where `response` is a dict representing the decoded JSON body.
+ """
+ logger.debug("get_json args: %s", args)
+
+ query_bytes = urllib.urlencode(args, True)
+ logger.debug("Query bytes: %s Retry DNS: %s", args, retry_on_dns_fail)
+
+ response = yield self._create_request(
+ destination.encode("ascii"),
+ "GET",
+ path.encode("ascii"),
+ query_bytes=query_bytes,
+ retry_on_dns_fail=retry_on_dns_fail,
+ body_callback=None
+ )
+
+ body = yield readBody(response)
+
+ defer.returnValue(json.loads(body))
+
+
+class MatrixFederationHttpClient(BaseHttpClient):
+ """HTTP client used to talk to other homeservers over the federation protocol.
+ Send client certificates and signs requests.
Attributes:
agent (twisted.web.client.Agent): The twisted Agent used to send the
requests.
"""
- RETRY_DNS_LOOKUP_FAILURES = "__retry_dns"
-
def __init__(self, hs):
self.signing_key = hs.config.signing_key[0]
self.server_name = hs.hostname
@@ -293,83 +358,17 @@ class MatrixHttpClient(BaseHttpClient):
)
-class IdentityServerHttpClient(BaseHttpClient):
- """Separate HTTP client for talking to the Identity servers since they
- don't use SRV records and talk x-www-form-urlencoded rather than JSON.
+class CaptchaServerHttpClient(BaseHttpClient):
+ """
+ Separate HTTP client for talking to google's captcha servers
+ Only slightly special because accepts partial download responses
"""
- def _getEndpoint(self, reactor, destination):
- #TODO: This should be talking TLS
- return matrix_endpoint(reactor, destination, timeout=10)
-
- @defer.inlineCallbacks
- def post_urlencoded_get_json(self, destination, path, args={}):
- logger.debug("post_urlencoded_get_json args: %s", args)
- query_bytes = urllib.urlencode(args, True)
-
- def body_callback(method, url_bytes, headers_dict):
- return FileBodyProducer(StringIO(query_bytes))
-
- response = yield self._create_request(
- destination.encode("ascii"),
- "POST",
- path.encode("ascii"),
- body_callback=body_callback,
- headers_dict={
- "Content-Type": ["application/x-www-form-urlencoded"]
- }
- )
-
- body = yield readBody(response)
-
- defer.returnValue(json.loads(body))
-
- @defer.inlineCallbacks
- def get_json(self, destination, path, args={}, retry_on_dns_fail=True):
- """ Get's some json from the given host homeserver and path
-
- Args:
- destination (str): The remote server to send the HTTP request
- to.
- path (str): The HTTP path.
- args (dict): A dictionary used to create query strings, defaults to
- None.
- **Note**: The value of each key is assumed to be an iterable
- and *not* a string.
-
- Returns:
- Deferred: Succeeds when we get *any* HTTP response.
-
- The result of the deferred is a tuple of `(code, response)`,
- where `response` is a dict representing the decoded JSON body.
- """
- logger.debug("get_json args: %s", args)
-
- query_bytes = urllib.urlencode(args, True)
- logger.debug("Query bytes: %s Retry DNS: %s", args, retry_on_dns_fail)
-
- response = yield self._create_request(
- destination.encode("ascii"),
- "GET",
- path.encode("ascii"),
- query_bytes=query_bytes,
- retry_on_dns_fail=retry_on_dns_fail,
- body_callback=None
- )
-
- body = yield readBody(response)
-
- defer.returnValue(json.loads(body))
-
-
-class CaptchaServerHttpClient(MatrixHttpClient):
- """Separate HTTP client for talking to google's captcha servers"""
def _getEndpoint(self, reactor, destination):
return matrix_endpoint(reactor, destination, timeout=10)
@defer.inlineCallbacks
- def post_urlencoded_get_raw(self, destination, path, accept_partial=False,
- args={}):
+ def post_urlencoded_get_raw(self, destination, path, args={}):
query_bytes = urllib.urlencode(args, True)
def body_callback(method, url_bytes, headers_dict):
@@ -389,10 +388,7 @@ class CaptchaServerHttpClient(MatrixHttpClient):
body = yield readBody(response)
defer.returnValue(body)
except PartialDownloadError as e:
- if accept_partial:
- defer.returnValue(e.response)
- else:
- raise e
+ defer.returnValue(e.response)
def _print_ex(e):
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..f3c3ca8191
--- /dev/null
+++ b/synapse/push/httppusher.py
@@ -0,0 +1,39 @@
+# -*- 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
+
+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
diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py
index e391e5678d..c38cf27690 100644
--- a/synapse/rest/__init__.py
+++ b/synapse/rest/__init__.py
@@ -16,7 +16,7 @@
from . import (
room, events, register, login, profile, presence, initial_sync, directory,
- voip, admin,
+ voip, admin, pusher,
)
@@ -45,3 +45,4 @@ class RestServletFactory(object):
directory.register_servlets(hs, client_resource)
voip.register_servlets(hs, client_resource)
admin.register_servlets(hs, client_resource)
+ pusher.register_servlets(hs, client_resource)
diff --git a/synapse/rest/pusher.py b/synapse/rest/pusher.py
new file mode 100644
index 0000000000..85d0d1c8cd
--- /dev/null
+++ b/synapse/rest/pusher.py
@@ -0,0 +1,71 @@
+# -*- 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.api.errors import SynapseError, Codes
+from synapse.push import PusherConfigException
+from base import RestServlet, client_path_pattern
+
+import json
+
+
+class PusherRestServlet(RestServlet):
+ PATTERN = client_path_pattern("/pushers/(?P<pushkey>[\w]*)$")
+
+ @defer.inlineCallbacks
+ def on_PUT(self, request, pushkey):
+ user = yield self.auth.get_user_by_req(request)
+
+ content = _parse_json(request)
+
+ reqd = ['kind', 'app', 'app_display_name', 'device_display_name', 'data']
+ missing = []
+ for i in reqd:
+ if i not in content:
+ missing.append(i)
+ if len(missing):
+ raise SynapseError(400, "Missing parameters: "+','.join(missing), errcode=Codes.MISSING_PARAM)
+
+ pusher_pool = self.hs.get_pusherpool()
+ try:
+ pusher_pool.add_pusher(user_name=user.to_string(),
+ kind=content['kind'],
+ app=content['app'],
+ app_display_name=content['app_display_name'],
+ device_display_name=content['device_display_name'],
+ pushkey=pushkey,
+ data=content['data'])
+ except PusherConfigException as pce:
+ raise SynapseError(400, "Config Error: "+pce.message, errcode=Codes.MISSING_PARAM)
+
+ defer.returnValue((200, {}))
+
+ def on_OPTIONS(self, request):
+ return (200, {})
+
+# XXX: C+ped from rest/room.py - surely this should be common?
+def _parse_json(request):
+ try:
+ content = json.loads(request.content.read())
+ if type(content) != dict:
+ raise SynapseError(400, "Content must be a JSON object.",
+ errcode=Codes.NOT_JSON)
+ return content
+ except ValueError:
+ raise SynapseError(400, "Content not JSON.", errcode=Codes.NOT_JSON)
+
+def register_servlets(hs, http_server):
+ PusherRestServlet(hs).register(http_server)
diff --git a/synapse/server.py b/synapse/server.py
index da0a44433a..cfbe7d5e38 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -36,6 +36,7 @@ from synapse.util.lockutils import LockManager
from synapse.streams.events import EventSources
from synapse.api.ratelimiting import Ratelimiter
from synapse.crypto.keyring import Keyring
+from synapse.push.pusherpool import PusherPool
class BaseHomeServer(object):
@@ -82,6 +83,7 @@ class BaseHomeServer(object):
'ratelimiter',
'keyring',
'event_validator',
+ 'pusherpool'
]
def __init__(self, hostname, **kwargs):
@@ -228,6 +230,9 @@ class HomeServer(BaseHomeServer):
def build_event_validator(self):
return EventValidator(self)
+ def build_pusherpool(self):
+ return PusherPool(self)
+
def register_servlets(self):
""" Register all servlets associated with this HomeServer.
"""
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index 1231794de0..e91fcc9789 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -33,6 +33,7 @@ from .stream import StreamStore
from .transactions import TransactionStore
from .keys import KeyStore
from .event_federation import EventFederationStore
+from .pusher import PusherStore
from .state import StateStore
from .signatures import SignatureStore
@@ -62,6 +63,7 @@ SCHEMAS = [
"state",
"event_edges",
"event_signatures",
+ "pusher"
]
@@ -81,7 +83,7 @@ class DataStore(RoomMemberStore, RoomStore,
RegistrationStore, StreamStore, ProfileStore, FeedbackStore,
PresenceStore, TransactionStore,
DirectoryStore, KeyStore, StateStore, SignatureStore,
- EventFederationStore, ):
+ EventFederationStore, PusherStore, ):
def __init__(self, hs):
super(DataStore, self).__init__(hs)
diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py
new file mode 100644
index 0000000000..047a5f42d9
--- /dev/null
+++ b/synapse/storage/pusher.py
@@ -0,0 +1,98 @@
+# -*- 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.
+
+import collections
+
+from ._base import SQLBaseStore, Table
+from twisted.internet import defer
+
+from sqlite3 import IntegrityError
+from synapse.api.errors import StoreError
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+class PusherStore(SQLBaseStore):
+ @defer.inlineCallbacks
+ def get_all_pushers_after_id(self, min_id):
+ sql = (
+ "SELECT id, user_name, kind, app, app_display_name, device_display_name, pushkey, data, last_token "
+ "FROM pushers "
+ "WHERE id > ?"
+ )
+
+ rows = yield self._execute(None, sql, min_id)
+
+ ret = [
+ {
+ "id": r[0],
+ "user_name": r[1],
+ "kind": r[2],
+ "app": r[3],
+ "app_display_name": r[4],
+ "device_display_name": r[5],
+ "pushkey": r[6],
+ "data": r[7],
+ "last_token": r[8]
+
+ }
+ for r in rows
+ ]
+
+ defer.returnValue(ret)
+
+ @defer.inlineCallbacks
+ def add_pusher(self, user_name, kind, app, app_display_name, device_display_name, pushkey, data):
+ try:
+ yield self._simple_insert(PushersTable.table_name, dict(
+ user_name=user_name,
+ kind=kind,
+ app=app,
+ app_display_name=app_display_name,
+ device_display_name=device_display_name,
+ pushkey=pushkey,
+ data=data
+ ))
+ except IntegrityError:
+ raise StoreError(409, "Pushkey in use.")
+ except Exception as e:
+ logger.error("create_pusher with failed: %s", e)
+ raise StoreError(500, "Problem creating pusher.")
+
+ @defer.inlineCallbacks
+ def update_pusher_last_token(self, user_name, pushkey, last_token):
+ yield self._simple_update_one(PushersTable.table_name,
+ {'user_name': user_name, 'pushkey': pushkey},
+ {'last_token': last_token}
+ )
+
+
+class PushersTable(Table):
+ table_name = "pushers"
+
+ fields = [
+ "id",
+ "user_name",
+ "kind",
+ "app"
+ "app_display_name",
+ "device_display_name",
+ "pushkey",
+ "data",
+ "last_token"
+ ]
+
+ EntryType = collections.namedtuple("PusherEntry", fields)
\ No newline at end of file
diff --git a/synapse/storage/schema/delta/v7.sql b/synapse/storage/schema/delta/v7.sql
new file mode 100644
index 0000000000..7f6852485d
--- /dev/null
+++ b/synapse/storage/schema/delta/v7.sql
@@ -0,0 +1,28 @@
+/* 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.
+ */
+-- Push notification endpoints that users have configured
+CREATE TABLE IF NOT EXISTS pushers (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_name TEXT NOT NULL,
+ kind varchar(8) NOT NULL,
+ app varchar(64) NOT NULL,
+ app_display_name varchar(64) NOT NULL,
+ device_display_name varchar(128) NOT NULL,
+ pushkey blob NOT NULL,
+ data text,
+ last_token TEXT,
+ FOREIGN KEY(user_name) REFERENCES users(name),
+ UNIQUE (user_name, pushkey)
+);
diff --git a/synapse/storage/schema/pusher.sql b/synapse/storage/schema/pusher.sql
new file mode 100644
index 0000000000..7f6852485d
--- /dev/null
+++ b/synapse/storage/schema/pusher.sql
@@ -0,0 +1,28 @@
+/* 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.
+ */
+-- Push notification endpoints that users have configured
+CREATE TABLE IF NOT EXISTS pushers (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_name TEXT NOT NULL,
+ kind varchar(8) NOT NULL,
+ app varchar(64) NOT NULL,
+ app_display_name varchar(64) NOT NULL,
+ device_display_name varchar(128) NOT NULL,
+ pushkey blob NOT NULL,
+ data text,
+ last_token TEXT,
+ FOREIGN KEY(user_name) REFERENCES users(name),
+ UNIQUE (user_name, pushkey)
+);
|
