diff --git a/contrib/jitsimeetbridge/jitsimeetbridge.py b/contrib/jitsimeetbridge/jitsimeetbridge.py
index dbc6f6ffa5..15f8e1c48b 100644
--- a/contrib/jitsimeetbridge/jitsimeetbridge.py
+++ b/contrib/jitsimeetbridge/jitsimeetbridge.py
@@ -39,43 +39,43 @@ ROOMDOMAIN="meet.jit.si"
#ROOMDOMAIN="conference.jitsi.vuc.me"
class TrivialMatrixClient:
- def __init__(self, access_token):
- self.token = None
- self.access_token = access_token
-
- def getEvent(self):
- while True:
- url = MATRIXBASE+'events?access_token='+self.access_token+"&timeout=60000"
- if self.token:
- url += "&from="+self.token
- req = grequests.get(url)
- resps = grequests.map([req])
- obj = json.loads(resps[0].content)
- print "incoming from matrix",obj
- if 'end' not in obj:
- continue
- self.token = obj['end']
- if len(obj['chunk']):
- return obj['chunk'][0]
-
- def joinRoom(self, roomId):
- url = MATRIXBASE+'rooms/'+roomId+'/join?access_token='+self.access_token
- print url
- headers={ 'Content-Type': 'application/json' }
- req = grequests.post(url, headers=headers, data='{}')
- resps = grequests.map([req])
- obj = json.loads(resps[0].content)
- print "response: ",obj
-
- def sendEvent(self, roomId, evType, event):
- url = MATRIXBASE+'rooms/'+roomId+'/send/'+evType+'?access_token='+self.access_token
- print url
- print json.dumps(event)
- headers={ 'Content-Type': 'application/json' }
- req = grequests.post(url, headers=headers, data=json.dumps(event))
- resps = grequests.map([req])
- obj = json.loads(resps[0].content)
- print "response: ",obj
+ def __init__(self, access_token):
+ self.token = None
+ self.access_token = access_token
+
+ def getEvent(self):
+ while True:
+ url = MATRIXBASE+'events?access_token='+self.access_token+"&timeout=60000"
+ if self.token:
+ url += "&from="+self.token
+ req = grequests.get(url)
+ resps = grequests.map([req])
+ obj = json.loads(resps[0].content)
+ print "incoming from matrix",obj
+ if 'end' not in obj:
+ continue
+ self.token = obj['end']
+ if len(obj['chunk']):
+ return obj['chunk'][0]
+
+ def joinRoom(self, roomId):
+ url = MATRIXBASE+'rooms/'+roomId+'/join?access_token='+self.access_token
+ print url
+ headers={ 'Content-Type': 'application/json' }
+ req = grequests.post(url, headers=headers, data='{}')
+ resps = grequests.map([req])
+ obj = json.loads(resps[0].content)
+ print "response: ",obj
+
+ def sendEvent(self, roomId, evType, event):
+ url = MATRIXBASE+'rooms/'+roomId+'/send/'+evType+'?access_token='+self.access_token
+ print url
+ print json.dumps(event)
+ headers={ 'Content-Type': 'application/json' }
+ req = grequests.post(url, headers=headers, data=json.dumps(event))
+ resps = grequests.map([req])
+ obj = json.loads(resps[0].content)
+ print "response: ",obj
@@ -83,178 +83,178 @@ xmppClients = {}
def matrixLoop():
- while True:
- ev = matrixCli.getEvent()
- print ev
- if ev['type'] == 'm.room.member':
- print 'membership event'
- if ev['membership'] == 'invite' and ev['state_key'] == MYUSERNAME:
- roomId = ev['room_id']
- print "joining room %s" % (roomId)
- matrixCli.joinRoom(roomId)
- elif ev['type'] == 'm.room.message':
- if ev['room_id'] in xmppClients:
- print "already have a bridge for that user, ignoring"
- continue
- print "got message, connecting"
- xmppClients[ev['room_id']] = TrivialXmppClient(ev['room_id'], ev['user_id'])
- gevent.spawn(xmppClients[ev['room_id']].xmppLoop)
- elif ev['type'] == 'm.call.invite':
- print "Incoming call"
- #sdp = ev['content']['offer']['sdp']
- #print "sdp: %s" % (sdp)
- #xmppClients[ev['room_id']] = TrivialXmppClient(ev['room_id'], ev['user_id'])
- #gevent.spawn(xmppClients[ev['room_id']].xmppLoop)
- elif ev['type'] == 'm.call.answer':
- print "Call answered"
- sdp = ev['content']['answer']['sdp']
- if ev['room_id'] not in xmppClients:
- print "We didn't have a call for that room"
- continue
- # should probably check call ID too
- xmppCli = xmppClients[ev['room_id']]
- xmppCli.sendAnswer(sdp)
- elif ev['type'] == 'm.call.hangup':
- if ev['room_id'] in xmppClients:
- xmppClients[ev['room_id']].stop()
- del xmppClients[ev['room_id']]
-
+ while True:
+ ev = matrixCli.getEvent()
+ print ev
+ if ev['type'] == 'm.room.member':
+ print 'membership event'
+ if ev['membership'] == 'invite' and ev['state_key'] == MYUSERNAME:
+ roomId = ev['room_id']
+ print "joining room %s" % (roomId)
+ matrixCli.joinRoom(roomId)
+ elif ev['type'] == 'm.room.message':
+ if ev['room_id'] in xmppClients:
+ print "already have a bridge for that user, ignoring"
+ continue
+ print "got message, connecting"
+ xmppClients[ev['room_id']] = TrivialXmppClient(ev['room_id'], ev['user_id'])
+ gevent.spawn(xmppClients[ev['room_id']].xmppLoop)
+ elif ev['type'] == 'm.call.invite':
+ print "Incoming call"
+ #sdp = ev['content']['offer']['sdp']
+ #print "sdp: %s" % (sdp)
+ #xmppClients[ev['room_id']] = TrivialXmppClient(ev['room_id'], ev['user_id'])
+ #gevent.spawn(xmppClients[ev['room_id']].xmppLoop)
+ elif ev['type'] == 'm.call.answer':
+ print "Call answered"
+ sdp = ev['content']['answer']['sdp']
+ if ev['room_id'] not in xmppClients:
+ print "We didn't have a call for that room"
+ continue
+ # should probably check call ID too
+ xmppCli = xmppClients[ev['room_id']]
+ xmppCli.sendAnswer(sdp)
+ elif ev['type'] == 'm.call.hangup':
+ if ev['room_id'] in xmppClients:
+ xmppClients[ev['room_id']].stop()
+ del xmppClients[ev['room_id']]
+
class TrivialXmppClient:
- def __init__(self, matrixRoom, userId):
- self.rid = 0
- self.matrixRoom = matrixRoom
- self.userId = userId
- self.running = True
-
- def stop(self):
- self.running = False
-
- def nextRid(self):
- self.rid += 1
- return '%d' % (self.rid)
-
- def sendIq(self, xml):
- fullXml = "<body rid='%s' xmlns='http://jabber.org/protocol/httpbind' sid='%s'>%s</body>" % (self.nextRid(), self.sid, xml)
- #print "\t>>>%s" % (fullXml)
- return self.xmppPoke(fullXml)
-
- def xmppPoke(self, xml):
- headers = {'Content-Type': 'application/xml'}
- req = grequests.post(HTTPBIND, verify=False, headers=headers, data=xml)
- resps = grequests.map([req])
- obj = BeautifulSoup(resps[0].content)
- return obj
-
- def sendAnswer(self, answer):
- print "sdp from matrix client",answer
- p = subprocess.Popen(['node', 'unjingle/unjingle.js', '--sdp'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
- jingle, out_err = p.communicate(answer)
- jingle = jingle % {
- 'tojid': self.callfrom,
- 'action': 'session-accept',
- 'initiator': self.callfrom,
- 'responder': self.jid,
- 'sid': self.callsid
- }
- print "answer jingle from sdp",jingle
- res = self.sendIq(jingle)
- print "reply from answer: ",res
-
- self.ssrcs = {}
- jingleSoup = BeautifulSoup(jingle)
- for cont in jingleSoup.iq.jingle.findAll('content'):
- if cont.description:
- self.ssrcs[cont['name']] = cont.description['ssrc']
- print "my ssrcs:",self.ssrcs
-
- gevent.joinall([
- gevent.spawn(self.advertiseSsrcs)
- ])
-
- def advertiseSsrcs(self):
+ def __init__(self, matrixRoom, userId):
+ self.rid = 0
+ self.matrixRoom = matrixRoom
+ self.userId = userId
+ self.running = True
+
+ def stop(self):
+ self.running = False
+
+ def nextRid(self):
+ self.rid += 1
+ return '%d' % (self.rid)
+
+ def sendIq(self, xml):
+ fullXml = "<body rid='%s' xmlns='http://jabber.org/protocol/httpbind' sid='%s'>%s</body>" % (self.nextRid(), self.sid, xml)
+ #print "\t>>>%s" % (fullXml)
+ return self.xmppPoke(fullXml)
+
+ def xmppPoke(self, xml):
+ headers = {'Content-Type': 'application/xml'}
+ req = grequests.post(HTTPBIND, verify=False, headers=headers, data=xml)
+ resps = grequests.map([req])
+ obj = BeautifulSoup(resps[0].content)
+ return obj
+
+ def sendAnswer(self, answer):
+ print "sdp from matrix client",answer
+ p = subprocess.Popen(['node', 'unjingle/unjingle.js', '--sdp'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+ jingle, out_err = p.communicate(answer)
+ jingle = jingle % {
+ 'tojid': self.callfrom,
+ 'action': 'session-accept',
+ 'initiator': self.callfrom,
+ 'responder': self.jid,
+ 'sid': self.callsid
+ }
+ print "answer jingle from sdp",jingle
+ res = self.sendIq(jingle)
+ print "reply from answer: ",res
+
+ self.ssrcs = {}
+ jingleSoup = BeautifulSoup(jingle)
+ for cont in jingleSoup.iq.jingle.findAll('content'):
+ if cont.description:
+ self.ssrcs[cont['name']] = cont.description['ssrc']
+ print "my ssrcs:",self.ssrcs
+
+ gevent.joinall([
+ gevent.spawn(self.advertiseSsrcs)
+ ])
+
+ def advertiseSsrcs(self):
time.sleep(7)
- print "SSRC spammer started"
- while self.running:
- ssrcMsg = "<presence to='%(tojid)s' xmlns='jabber:client'><x xmlns='http://jabber.org/protocol/muc'/><c xmlns='http://jabber.org/protocol/caps' hash='sha-1' node='http://jitsi.org/jitsimeet' ver='0WkSdhFnAUxrz4ImQQLdB80GFlE='/><nick xmlns='http://jabber.org/protocol/nick'>%(nick)s</nick><stats xmlns='http://jitsi.org/jitmeet/stats'><stat name='bitrate_download' value='175'/><stat name='bitrate_upload' value='176'/><stat name='packetLoss_total' value='0'/><stat name='packetLoss_download' value='0'/><stat name='packetLoss_upload' value='0'/></stats><media xmlns='http://estos.de/ns/mjs'><source type='audio' ssrc='%(assrc)s' direction='sendre'/><source type='video' ssrc='%(vssrc)s' direction='sendre'/></media></presence>" % { 'tojid': "%s@%s/%s" % (ROOMNAME, ROOMDOMAIN, self.shortJid), 'nick': self.userId, 'assrc': self.ssrcs['audio'], 'vssrc': self.ssrcs['video'] }
- res = self.sendIq(ssrcMsg)
- print "reply from ssrc announce: ",res
- time.sleep(10)
-
-
-
- def xmppLoop(self):
- self.matrixCallId = time.time()
- res = self.xmppPoke("<body rid='%s' xmlns='http://jabber.org/protocol/httpbind' to='%s' xml:lang='en' wait='60' hold='1' content='text/xml; charset=utf-8' ver='1.6' xmpp:version='1.0' xmlns:xmpp='urn:xmpp:xbosh'/>" % (self.nextRid(), HOST))
-
- print res
- self.sid = res.body['sid']
- print "sid %s" % (self.sid)
-
- res = self.sendIq("<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='ANONYMOUS'/>")
-
- res = self.xmppPoke("<body rid='%s' xmlns='http://jabber.org/protocol/httpbind' sid='%s' to='%s' xml:lang='en' xmpp:restart='true' xmlns:xmpp='urn:xmpp:xbosh'/>" % (self.nextRid(), self.sid, HOST))
-
- res = self.sendIq("<iq type='set' id='_bind_auth_2' xmlns='jabber:client'><bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/></iq>")
- print res
-
- self.jid = res.body.iq.bind.jid.string
- print "jid: %s" % (self.jid)
- self.shortJid = self.jid.split('-')[0]
-
- res = self.sendIq("<iq type='set' id='_session_auth_2' xmlns='jabber:client'><session xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>")
-
- #randomthing = res.body.iq['to']
- #whatsitpart = randomthing.split('-')[0]
-
- #print "other random bind thing: %s" % (randomthing)
-
- # advertise preence to the jitsi room, with our nick
- res = self.sendIq("<iq type='get' to='%s' xmlns='jabber:client' id='1:sendIQ'><services xmlns='urn:xmpp:extdisco:1'><service host='%s'/></services></iq><presence to='%s@%s/d98f6c40' xmlns='jabber:client'><x xmlns='http://jabber.org/protocol/muc'/><c xmlns='http://jabber.org/protocol/caps' hash='sha-1' node='http://jitsi.org/jitsimeet' ver='0WkSdhFnAUxrz4ImQQLdB80GFlE='/><nick xmlns='http://jabber.org/protocol/nick'>%s</nick></presence>" % (HOST, TURNSERVER, ROOMNAME, ROOMDOMAIN, self.userId))
- self.muc = {'users': []}
- for p in res.body.findAll('presence'):
- u = {}
- u['shortJid'] = p['from'].split('/')[1]
- if p.c and p.c.nick:
- u['nick'] = p.c.nick.string
- self.muc['users'].append(u)
- print "muc: ",self.muc
-
- # wait for stuff
- while True:
- print "waiting..."
- res = self.sendIq("")
- print "got from stream: ",res
- if res.body.iq:
- jingles = res.body.iq.findAll('jingle')
- if len(jingles):
- self.callfrom = res.body.iq['from']
- self.handleInvite(jingles[0])
- elif 'type' in res.body and res.body['type'] == 'terminate':
- self.running = False
- del xmppClients[self.matrixRoom]
- return
-
- def handleInvite(self, jingle):
- self.initiator = jingle['initiator']
- self.callsid = jingle['sid']
- p = subprocess.Popen(['node', 'unjingle/unjingle.js', '--jingle'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
- print "raw jingle invite",str(jingle)
- sdp, out_err = p.communicate(str(jingle))
- print "transformed remote offer sdp",sdp
- inviteEvent = {
- 'offer': {
- 'type': 'offer',
- 'sdp': sdp
- },
- 'call_id': self.matrixCallId,
- 'version': 0,
- 'lifetime': 30000
- }
- matrixCli.sendEvent(self.matrixRoom, 'm.call.invite', inviteEvent)
-
+ print "SSRC spammer started"
+ while self.running:
+ ssrcMsg = "<presence to='%(tojid)s' xmlns='jabber:client'><x xmlns='http://jabber.org/protocol/muc'/><c xmlns='http://jabber.org/protocol/caps' hash='sha-1' node='http://jitsi.org/jitsimeet' ver='0WkSdhFnAUxrz4ImQQLdB80GFlE='/><nick xmlns='http://jabber.org/protocol/nick'>%(nick)s</nick><stats xmlns='http://jitsi.org/jitmeet/stats'><stat name='bitrate_download' value='175'/><stat name='bitrate_upload' value='176'/><stat name='packetLoss_total' value='0'/><stat name='packetLoss_download' value='0'/><stat name='packetLoss_upload' value='0'/></stats><media xmlns='http://estos.de/ns/mjs'><source type='audio' ssrc='%(assrc)s' direction='sendre'/><source type='video' ssrc='%(vssrc)s' direction='sendre'/></media></presence>" % { 'tojid': "%s@%s/%s" % (ROOMNAME, ROOMDOMAIN, self.shortJid), 'nick': self.userId, 'assrc': self.ssrcs['audio'], 'vssrc': self.ssrcs['video'] }
+ res = self.sendIq(ssrcMsg)
+ print "reply from ssrc announce: ",res
+ time.sleep(10)
+
+
+
+ def xmppLoop(self):
+ self.matrixCallId = time.time()
+ res = self.xmppPoke("<body rid='%s' xmlns='http://jabber.org/protocol/httpbind' to='%s' xml:lang='en' wait='60' hold='1' content='text/xml; charset=utf-8' ver='1.6' xmpp:version='1.0' xmlns:xmpp='urn:xmpp:xbosh'/>" % (self.nextRid(), HOST))
+
+ print res
+ self.sid = res.body['sid']
+ print "sid %s" % (self.sid)
+
+ res = self.sendIq("<auth xmlns='urn:ietf:params:xml:ns:xmpp-sasl' mechanism='ANONYMOUS'/>")
+
+ res = self.xmppPoke("<body rid='%s' xmlns='http://jabber.org/protocol/httpbind' sid='%s' to='%s' xml:lang='en' xmpp:restart='true' xmlns:xmpp='urn:xmpp:xbosh'/>" % (self.nextRid(), self.sid, HOST))
+
+ res = self.sendIq("<iq type='set' id='_bind_auth_2' xmlns='jabber:client'><bind xmlns='urn:ietf:params:xml:ns:xmpp-bind'/></iq>")
+ print res
+
+ self.jid = res.body.iq.bind.jid.string
+ print "jid: %s" % (self.jid)
+ self.shortJid = self.jid.split('-')[0]
+
+ res = self.sendIq("<iq type='set' id='_session_auth_2' xmlns='jabber:client'><session xmlns='urn:ietf:params:xml:ns:xmpp-session'/></iq>")
+
+ #randomthing = res.body.iq['to']
+ #whatsitpart = randomthing.split('-')[0]
+
+ #print "other random bind thing: %s" % (randomthing)
+
+ # advertise preence to the jitsi room, with our nick
+ res = self.sendIq("<iq type='get' to='%s' xmlns='jabber:client' id='1:sendIQ'><services xmlns='urn:xmpp:extdisco:1'><service host='%s'/></services></iq><presence to='%s@%s/d98f6c40' xmlns='jabber:client'><x xmlns='http://jabber.org/protocol/muc'/><c xmlns='http://jabber.org/protocol/caps' hash='sha-1' node='http://jitsi.org/jitsimeet' ver='0WkSdhFnAUxrz4ImQQLdB80GFlE='/><nick xmlns='http://jabber.org/protocol/nick'>%s</nick></presence>" % (HOST, TURNSERVER, ROOMNAME, ROOMDOMAIN, self.userId))
+ self.muc = {'users': []}
+ for p in res.body.findAll('presence'):
+ u = {}
+ u['shortJid'] = p['from'].split('/')[1]
+ if p.c and p.c.nick:
+ u['nick'] = p.c.nick.string
+ self.muc['users'].append(u)
+ print "muc: ",self.muc
+
+ # wait for stuff
+ while True:
+ print "waiting..."
+ res = self.sendIq("")
+ print "got from stream: ",res
+ if res.body.iq:
+ jingles = res.body.iq.findAll('jingle')
+ if len(jingles):
+ self.callfrom = res.body.iq['from']
+ self.handleInvite(jingles[0])
+ elif 'type' in res.body and res.body['type'] == 'terminate':
+ self.running = False
+ del xmppClients[self.matrixRoom]
+ return
+
+ def handleInvite(self, jingle):
+ self.initiator = jingle['initiator']
+ self.callsid = jingle['sid']
+ p = subprocess.Popen(['node', 'unjingle/unjingle.js', '--jingle'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+ print "raw jingle invite",str(jingle)
+ sdp, out_err = p.communicate(str(jingle))
+ print "transformed remote offer sdp",sdp
+ inviteEvent = {
+ 'offer': {
+ 'type': 'offer',
+ 'sdp': sdp
+ },
+ 'call_id': self.matrixCallId,
+ 'version': 0,
+ 'lifetime': 30000
+ }
+ matrixCli.sendEvent(self.matrixRoom, 'm.call.invite', inviteEvent)
+
matrixCli = TrivialMatrixClient(ACCESS_TOKEN)
gevent.joinall([
- gevent.spawn(matrixLoop)
+ gevent.spawn(matrixLoop)
])
diff --git a/synapse/api/errors.py b/synapse/api/errors.py
index 2b049debf3..55181fe77e 100644
--- a/synapse/api/errors.py
+++ b/synapse/api/errors.py
@@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
class Codes(object):
+ UNRECOGNIZED = "M_UNRECOGNIZED"
UNAUTHORIZED = "M_UNAUTHORIZED"
FORBIDDEN = "M_FORBIDDEN"
BAD_JSON = "M_BAD_JSON"
@@ -34,6 +35,7 @@ class Codes(object):
LIMIT_EXCEEDED = "M_LIMIT_EXCEEDED"
CAPTCHA_NEEDED = "M_CAPTCHA_NEEDED"
CAPTCHA_INVALID = "M_CAPTCHA_INVALID"
+ MISSING_PARAM = "M_MISSING_PARAM",
TOO_LARGE = "M_TOO_LARGE"
@@ -81,6 +83,17 @@ class RegistrationError(SynapseError):
pass
+class UnrecognizedRequestError(SynapseError):
+ """An error indicating we don't understand the request you're trying to make"""
+ def __init__(self, *args, **kwargs):
+ if "errcode" not in kwargs:
+ kwargs["errcode"] = Codes.NOT_FOUND
+ super(UnrecognizedRequestError, self).__init__(
+ 400,
+ "Unrecognized request",
+ **kwargs
+ )
+
class AuthError(SynapseError):
"""An error raised when there was a problem authorising an event."""
diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py
index cd24bbdc79..6ebf1f0de0 100755
--- a/synapse/app/homeserver.py
+++ b/synapse/app/homeserver.py
@@ -268,6 +268,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/events.py b/synapse/handlers/events.py
index 103bc67c42..75fb941008 100644
--- a/synapse/handlers/events.py
+++ b/synapse/handlers/events.py
@@ -47,11 +47,11 @@ class EventStreamHandler(BaseHandler):
@defer.inlineCallbacks
@log_function
def get_stream(self, auth_user_id, pagin_config, timeout=0,
- as_client_event=True):
+ as_client_event=True, affect_presence=True):
auth_user = self.hs.parse_userid(auth_user_id)
try:
- if auth_user not in self._streams_per_user:
+ if affect_presence and auth_user not in self._streams_per_user:
self._streams_per_user[auth_user] = 0
if auth_user in self._stop_timer_per_user:
try:
@@ -64,7 +64,7 @@ class EventStreamHandler(BaseHandler):
yield self.distributor.fire(
"started_user_eventstream", auth_user
)
- self._streams_per_user[auth_user] += 1
+ self._streams_per_user[auth_user] += 1
if pagin_config.from_token is None:
pagin_config.from_token = None
@@ -90,27 +90,28 @@ class EventStreamHandler(BaseHandler):
defer.returnValue(chunk)
finally:
- self._streams_per_user[auth_user] -= 1
- if not self._streams_per_user[auth_user]:
- del self._streams_per_user[auth_user]
-
- # 10 seconds of grace to allow the client to reconnect again
- # before we think they're gone
- def _later():
- logger.debug(
- "_later stopped_user_eventstream %s", auth_user
- )
+ if affect_presence:
+ self._streams_per_user[auth_user] -= 1
+ if not self._streams_per_user[auth_user]:
+ del self._streams_per_user[auth_user]
+
+ # 10 seconds of grace to allow the client to reconnect again
+ # before we think they're gone
+ def _later():
+ logger.debug(
+ "_later stopped_user_eventstream %s", auth_user
+ )
- self._stop_timer_per_user.pop(auth_user, None)
+ self._stop_timer_per_user.pop(auth_user, None)
- yield self.distributor.fire(
- "stopped_user_eventstream", auth_user
- )
+ yield self.distributor.fire(
+ "stopped_user_eventstream", auth_user
+ )
- logger.debug("Scheduling _later: for %s", auth_user)
- self._stop_timer_per_user[auth_user] = (
- self.clock.call_later(30, _later)
- )
+ logger.debug("Scheduling _later: for %s", auth_user)
+ self._stop_timer_per_user[auth_user] = (
+ self.clock.call_later(30, _later)
+ )
class EventHandler(BaseHandler):
diff --git a/synapse/http/client.py b/synapse/http/client.py
index 7793bab106..198f575cfa 100644
--- a/synapse/http/client.py
+++ b/synapse/http/client.py
@@ -63,6 +63,25 @@ class SimpleHttpClient(object):
defer.returnValue(json.loads(body))
@defer.inlineCallbacks
+ def post_json_get_json(self, uri, post_json):
+ json_str = json.dumps(post_json)
+
+ logger.info("HTTP POST %s -> %s", json_str, uri)
+
+ response = yield self.agent.request(
+ "POST",
+ uri.encode("ascii"),
+ headers=Headers({
+ "Content-Type": ["application/json"]
+ }),
+ bodyProducer=FileBodyProducer(StringIO(json_str))
+ )
+
+ body = yield readBody(response)
+
+ defer.returnValue(json.loads(body))
+
+ @defer.inlineCallbacks
def get_json(self, uri, args={}):
""" Get's some json from the given host and path
diff --git a/synapse/http/server.py b/synapse/http/server.py
index 8015a22edf..0f6539e1be 100644
--- a/synapse/http/server.py
+++ b/synapse/http/server.py
@@ -16,7 +16,7 @@
from synapse.http.agent_name import AGENT_NAME
from synapse.api.errors import (
- cs_exception, SynapseError, CodeMessageException
+ cs_exception, SynapseError, CodeMessageException, UnrecognizedRequestError
)
from synapse.util.logcontext import LoggingContext
@@ -139,11 +139,7 @@ class JsonResource(HttpServer, resource.Resource):
return
# Huh. No one wanted to handle that? Fiiiiiine. Send 400.
- self._send_response(
- request,
- 400,
- {"error": "Unrecognized request"}
- )
+ raise UnrecognizedRequestError()
except CodeMessageException as e:
if isinstance(e, SynapseError):
logger.info("%s SynapseError: %s - %s", request, e.code, e.msg)
diff --git a/synapse/push/__init__.py b/synapse/push/__init__.py
new file mode 100644
index 0000000000..3ee652f3bc
--- /dev/null
+++ b/synapse/push/__init__.py
@@ -0,0 +1,220 @@
+# -*- 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 = 60 * 60 * 1000
+ GIVE_UP_AFTER = 24 * 60 * 60 * 1000
+
+ def __init__(self, _hs, instance_handle, user_name, app_id,
+ app_display_name, device_display_name, pushkey, pushkey_ts,
+ data, last_token, last_success, failing_since):
+ self.hs = _hs
+ self.evStreamHandler = self.hs.get_handlers().event_stream_handler
+ self.store = self.hs.get_datastore()
+ self.clock = self.hs.get_clock()
+ self.instance_handle = instance_handle,
+ self.user_name = user_name
+ self.app_id = app_id
+ self.app_display_name = app_display_name
+ self.device_display_name = device_display_name
+ self.pushkey = pushkey
+ self.pushkey_ts = pushkey_ts
+ self.data = data
+ self.last_token = last_token
+ self.last_success = last_success # not actually used
+ self.backoff_delay = Pusher.INITIAL_BACKOFF
+ self.failing_since = failing_since
+ self.alive = True
+
+ def _should_notify_for_event(self, ev):
+ """
+ This should take into account notification settings that the user
+ has configured both globally and per-room when we have the ability
+ to do such things.
+ """
+ if ev['user_id'] == self.user_name:
+ # let's assume you probably know about messages you sent yourself
+ return False
+ return True
+
+ @defer.inlineCallbacks
+ def get_context_for_event(self, ev):
+ name_aliases = yield self.store.get_room_name_and_aliases(
+ ev['room_id']
+ )
+
+ ctx = {'aliases': name_aliases[1]}
+ if name_aliases[0] is not None:
+ ctx['name'] = name_aliases[0]
+
+ defer.returnValue(ctx)
+
+
+ @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 self.alive:
+ 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, affect_presence=False
+ )
+
+ # limiting to 1 may get 1 event plus 1 presence event, so
+ # pick out the actual event
+ single_event = None
+ for c in chunk['chunk']:
+ if 'event_id' in c: # Hmmm...
+ single_event = c
+ break
+ if not single_event:
+ self.last_token = chunk['end']
+ continue
+
+ if not self.alive:
+ continue
+
+ processed = False
+ if self._should_notify_for_event(single_event):
+ rejected = yield self.dispatch_push(single_event)
+ if not rejected == False:
+ processed = True
+ for pk in rejected:
+ if pk != self.pushkey:
+ # for sanity, we only remove the pushkey if it
+ # was the one we actually sent...
+ logger.warn(
+ ("Ignoring rejected pushkey %s because we" +
+ "didn't send it"), (pk,)
+ )
+ else:
+ logger.info(
+ "Pushkey %s was rejected: removing",
+ pk
+ )
+ yield self.hs.get_pusherpool().remove_pusher(
+ self.app_id, pk
+ )
+ else:
+ processed = True
+
+ if not self.alive:
+ continue
+
+ if processed:
+ self.backoff_delay = Pusher.INITIAL_BACKOFF
+ self.last_token = chunk['end']
+ self.store.update_pusher_last_token_and_success(
+ self.user_name,
+ self.pushkey,
+ self.last_token,
+ self.clock.time_msec()
+ )
+ if self.failing_since:
+ self.failing_since = None
+ self.store.update_pusher_failing_since(
+ self.user_name,
+ self.pushkey,
+ self.failing_since)
+ else:
+ if not self.failing_since:
+ self.failing_since = self.clock.time_msec()
+ self.store.update_pusher_failing_since(
+ self.user_name,
+ self.pushkey,
+ self.failing_since
+ )
+
+ if self.failing_since and \
+ self.failing_since < \
+ self.clock.time_msec() - Pusher.GIVE_UP_AFTER:
+ # we really only give up so that if the URL gets
+ # fixed, we don't suddenly deliver a load
+ # of old notifications.
+ logger.warn("Giving up on a notification to user %s, "
+ "pushkey %s",
+ self.user_name, self.pushkey)
+ 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
+ )
+
+ self.failing_since = None
+ self.store.update_pusher_failing_since(
+ self.user_name,
+ self.pushkey,
+ self.failing_since
+ )
+ else:
+ logger.warn("Failed to dispatch push for user %s "
+ "(failing for %dms)."
+ "Trying again in %dms",
+ self.user_name,
+ self.clock.time_msec() - self.failing_since,
+ 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
+
+ def stop(self):
+ self.alive = False
+
+ def dispatch_push(self, p):
+ """
+ Overridden by implementing classes to actually deliver the notification
+ :param p: The event to notify for as a single event from the event stream
+ :return: If the notification was delivered, an array containing any
+ pushkeys that were rejected by the push gateway.
+ False if the notification could not be delivered (ie.
+ should be retried).
+ """
+ pass
+
+
+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..46433ad4a9
--- /dev/null
+++ b/synapse/push/httppusher.py
@@ -0,0 +1,112 @@
+# -*- 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 SimpleHttpClient
+
+from twisted.internet import defer
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class HttpPusher(Pusher):
+ def __init__(self, _hs, instance_handle, user_name, app_id,
+ app_display_name, device_display_name, pushkey, pushkey_ts,
+ data, last_token, last_success, failing_since):
+ super(HttpPusher, self).__init__(
+ _hs,
+ instance_handle,
+ user_name,
+ app_id,
+ app_display_name,
+ device_display_name,
+ pushkey,
+ pushkey_ts,
+ data,
+ last_token,
+ last_success,
+ failing_since
+ )
+ if 'url' not in data:
+ raise PusherConfigException(
+ "'url' required in data for HTTP pusher"
+ )
+ self.url = data['url']
+ self.httpCli = SimpleHttpClient(self.hs)
+ self.data_minus_url = {}
+ self.data_minus_url.update(self.data)
+ del self.data_minus_url['url']
+
+ @defer.inlineCallbacks
+ def _build_notification_dict(self, event):
+ # we probably do not want to push for every presence update
+ # (we may want to be able to set up notifications when specific
+ # people sign in, but we'd want to only deliver the pertinent ones)
+ # Actually, presence events will not get this far now because we
+ # need to filter them out in the main Pusher code.
+ if 'event_id' not in event:
+ defer.returnValue(None)
+
+ ctx = yield self.get_context_for_event(event)
+
+ d = {
+ 'notification': {
+ 'transition': 'new',
+ # everything is new for now: we don't have read receipts
+ 'id': event['event_id'],
+ 'type': event['type'],
+ 'from': event['user_id'],
+ # we may have to fetch this over federation and we
+ # can't trust it anyway: is it worth it?
+ #'from_display_name': 'Steve Stevington'
+ #'counts': { -- we don't mark messages as read yet so
+ # we have no way of knowing
+ # 'unread': 1,
+ # 'missed_calls': 2
+ # },
+ 'devices': [
+ {
+ 'app_id': self.app_id,
+ 'pushkey': self.pushkey,
+ 'pushkey_ts': long(self.pushkey_ts / 1000),
+ 'data': self.data_minus_url
+ }
+ ]
+ }
+ }
+
+ if len(ctx['aliases']):
+ d['notification']['roomAlias'] = ctx['aliases'][0]
+ if 'name' in ctx:
+ d['notification']['roomName'] = ctx['name']
+
+ defer.returnValue(d)
+
+ @defer.inlineCallbacks
+ def dispatch_push(self, event):
+ notification_dict = yield self._build_notification_dict(event)
+ if not notification_dict:
+ defer.returnValue([])
+ try:
+ resp = yield self.httpCli.post_json_get_json(self.url, notification_dict)
+ except:
+ logger.exception("Failed to push %s ", self.url)
+ defer.returnValue(False)
+ rejected = []
+ if 'rejected' in resp:
+ rejected = resp['rejected']
+ defer.returnValue(rejected)
diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py
new file mode 100644
index 0000000000..2dfecf178b
--- /dev/null
+++ b/synapse/push/pusherpool.py
@@ -0,0 +1,137 @@
+#!/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
+
+ @defer.inlineCallbacks
+ def start(self):
+ pushers = yield self.store.get_all_pushers()
+ for p in pushers:
+ p['data'] = json.loads(p['data'])
+ self._start_pushers(pushers)
+
+ @defer.inlineCallbacks
+ def add_pusher(self, user_name, instance_handle, kind, app_id,
+ app_display_name, device_display_name, pushkey, lang, 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,
+ "instance_handle": instance_handle,
+ "app_id": app_id,
+ "app_display_name": app_display_name,
+ "device_display_name": device_display_name,
+ "pushkey": pushkey,
+ "pushkey_ts": self.hs.get_clock().time_msec(),
+ "lang": lang,
+ "data": data,
+ "last_token": None,
+ "last_success": None,
+ "failing_since": None
+ })
+ yield self._add_pusher_to_store(
+ user_name, instance_handle, kind, app_id,
+ app_display_name, device_display_name,
+ pushkey, lang, data
+ )
+
+ @defer.inlineCallbacks
+ def _add_pusher_to_store(self, user_name, instance_handle, kind, app_id,
+ app_display_name, device_display_name,
+ pushkey, lang, data):
+ yield self.store.add_pusher(
+ user_name=user_name,
+ instance_handle=instance_handle,
+ kind=kind,
+ app_id=app_id,
+ app_display_name=app_display_name,
+ device_display_name=device_display_name,
+ pushkey=pushkey,
+ pushkey_ts=self.hs.get_clock().time_msec(),
+ lang=lang,
+ data=json.dumps(data)
+ )
+ self._refresh_pusher((app_id, pushkey))
+
+ def _create_pusher(self, pusherdict):
+ if pusherdict['kind'] == 'http':
+ return HttpPusher(
+ self.hs,
+ instance_handle=pusherdict['instance_handle'],
+ user_name=pusherdict['user_name'],
+ app_id=pusherdict['app_id'],
+ app_display_name=pusherdict['app_display_name'],
+ device_display_name=pusherdict['device_display_name'],
+ pushkey=pusherdict['pushkey'],
+ pushkey_ts=pusherdict['pushkey_ts'],
+ data=pusherdict['data'],
+ last_token=pusherdict['last_token'],
+ last_success=pusherdict['last_success'],
+ failing_since=pusherdict['failing_since']
+ )
+ else:
+ raise PusherConfigException(
+ "Unknown pusher type '%s' for user %s" %
+ (pusherdict['kind'], pusherdict['user_name'])
+ )
+
+ @defer.inlineCallbacks
+ def _refresh_pusher(self, app_id_pushkey):
+ p = yield self.store.get_pushers_by_app_id_and_pushkey(
+ app_id_pushkey
+ )
+ p['data'] = json.loads(p['data'])
+
+ self._start_pushers([p])
+
+ def _start_pushers(self, pushers):
+ logger.info("Starting %d pushers", len(pushers))
+ for pusherdict in pushers:
+ p = self._create_pusher(pusherdict)
+ if p:
+ fullid = "%s:%s" % (pusherdict['app_id'], pusherdict['pushkey'])
+ if fullid in self.pushers:
+ self.pushers[fullid].stop()
+ self.pushers[fullid] = p
+ p.start()
+
+ @defer.inlineCallbacks
+ def remove_pusher(self, app_id, pushkey):
+ fullid = "%s:%s" % (app_id, pushkey)
+ if fullid in self.pushers:
+ logger.info("Stopping pusher %s", fullid)
+ self.pushers[fullid].stop()
+ del self.pushers[fullid]
+ yield self.store.delete_pusher_by_app_id_pushkey(app_id, pushkey)
\ No newline at end of file
diff --git a/synapse/rest/__init__.py b/synapse/rest/__init__.py
index 1a84d94cd9..90afd93333 100644
--- a/synapse/rest/__init__.py
+++ b/synapse/rest/__init__.py
@@ -11,4 +11,4 @@
# 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.
+# limitations under the License.
\ No newline at end of file
diff --git a/synapse/rest/client/v1/__init__.py b/synapse/rest/client/v1/__init__.py
index 88ec9cd27d..8e5877cf3f 100644
--- a/synapse/rest/client/v1/__init__.py
+++ b/synapse/rest/client/v1/__init__.py
@@ -16,7 +16,7 @@
from . import (
room, events, register, login, profile, presence, initial_sync, directory,
- voip, admin,
+ voip, admin, pusher, push_rule
)
@@ -45,3 +45,5 @@ 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)
+ push_rule.register_servlets(hs, client_resource)
diff --git a/synapse/rest/client/v1/push_rule.py b/synapse/rest/client/v1/push_rule.py
new file mode 100644
index 0000000000..b5e74479cf
--- /dev/null
+++ b/synapse/rest/client/v1/push_rule.py
@@ -0,0 +1,195 @@
+# -*- 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, UnrecognizedRequestError
+from base import RestServlet, client_path_pattern
+from synapse.storage.push_rule import InconsistentRuleException, RuleNotFoundException
+
+import json
+
+
+class PushRuleRestServlet(RestServlet):
+ PATTERN = client_path_pattern("/pushrules/.*$")
+
+ def rule_spec_from_path(self, path):
+ if len(path) < 2:
+ raise UnrecognizedRequestError()
+ if path[0] != 'pushrules':
+ raise UnrecognizedRequestError()
+
+ scope = path[1]
+ path = path[2:]
+ if scope not in ['global', 'device']:
+ raise UnrecognizedRequestError()
+
+ device = None
+ if scope == 'device':
+ if len(path) == 0:
+ raise UnrecognizedRequestError()
+ device = path[0]
+ path = path[1:]
+
+ if len(path) == 0:
+ raise UnrecognizedRequestError()
+
+ template = path[0]
+ path = path[1:]
+
+ if len(path) == 0:
+ raise UnrecognizedRequestError()
+
+ rule_id = path[0]
+
+ spec = {
+ 'scope' : scope,
+ 'template': template,
+ 'rule_id': rule_id
+ }
+ if device:
+ spec['device'] = device
+ return spec
+
+ def rule_tuple_from_request_object(self, rule_template, rule_id, req_obj):
+ if rule_template in ['override', 'underride']:
+ if 'conditions' not in req_obj:
+ raise InvalidRuleException("Missing 'conditions'")
+ conditions = req_obj['conditions']
+ for c in conditions:
+ if 'kind' not in c:
+ raise InvalidRuleException("Condition without 'kind'")
+ elif rule_template == 'room':
+ conditions = [{
+ 'kind': 'event_match',
+ 'key': 'room_id',
+ 'pattern': rule_id
+ }]
+ elif rule_template == 'sender':
+ conditions = [{
+ 'kind': 'event_match',
+ 'key': 'user_id',
+ 'pattern': rule_id
+ }]
+ elif rule_template == 'content':
+ if 'pattern' not in req_obj:
+ raise InvalidRuleException("Content rule missing 'pattern'")
+ conditions = [{
+ 'kind': 'event_match',
+ 'key': 'content.body',
+ 'pattern': req_obj['pattern']
+ }]
+ else:
+ raise InvalidRuleException("Unknown rule template: %s" % (rule_template))
+
+ if 'actions' not in req_obj:
+ raise InvalidRuleException("No actions found")
+ actions = req_obj['actions']
+
+ for a in actions:
+ if a in ['notify', 'dont-notify', 'coalesce']:
+ pass
+ elif isinstance(a, dict) and 'set_sound' in a:
+ pass
+ else:
+ raise InvalidRuleException("Unrecognised action")
+
+ return (conditions, actions)
+
+ def priority_class_from_spec(self, spec):
+ map = {
+ 'underride': 0,
+ 'sender': 1,
+ 'room': 2,
+ 'content': 3,
+ 'override': 4
+ }
+
+ if spec['template'] not in map.keys():
+ raise InvalidRuleException("Unknown template: %s" % (spec['kind']))
+ pc = map[spec['template']]
+
+ if spec['scope'] == 'device':
+ pc += 5
+
+ return pc
+
+ @defer.inlineCallbacks
+ def on_PUT(self, request):
+ spec = self.rule_spec_from_path(request.postpath)
+ try:
+ priority_class = self.priority_class_from_spec(spec)
+ except InvalidRuleException as e:
+ raise SynapseError(400, e.message)
+
+ user = yield self.auth.get_user_by_req(request)
+
+ content = _parse_json(request)
+
+ try:
+ (conditions, actions) = self.rule_tuple_from_request_object(
+ spec['template'],
+ spec['rule_id'],
+ content
+ )
+ except InvalidRuleException as e:
+ raise SynapseError(400, e.message)
+
+ before = request.args.get("before", None)
+ if before and len(before):
+ before = before[0]
+ after = request.args.get("after", None)
+ if after and len(after):
+ after = after[0]
+
+ try:
+ yield self.hs.get_datastore().add_push_rule(
+ user_name=user.to_string(),
+ rule_id=spec['rule_id'],
+ priority_class=priority_class,
+ conditions=conditions,
+ actions=actions,
+ before=before,
+ after=after
+ )
+ except InconsistentRuleException as e:
+ raise SynapseError(400, e.message)
+ except RuleNotFoundException:
+ raise SynapseError(400, "before/after rule not found")
+
+ defer.returnValue((200, {}))
+
+ def on_OPTIONS(self, _):
+ return 200, {}
+
+
+class InvalidRuleException(Exception):
+ pass
+
+
+# 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):
+ PushRuleRestServlet(hs).register(http_server)
diff --git a/synapse/rest/client/v1/pusher.py b/synapse/rest/client/v1/pusher.py
new file mode 100644
index 0000000000..4659c9b1d9
--- /dev/null
+++ b/synapse/rest/client/v1/pusher.py
@@ -0,0 +1,80 @@
+# -*- 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/set$")
+
+ @defer.inlineCallbacks
+ def on_POST(self, request):
+ user = yield self.auth.get_user_by_req(request)
+
+ content = _parse_json(request)
+
+ reqd = ['instance_handle', 'kind', 'app_id', 'app_display_name',
+ 'device_display_name', 'pushkey', 'lang', '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:
+ yield pusher_pool.add_pusher(
+ user_name=user.to_string(),
+ instance_handle=content['instance_handle'],
+ kind=content['kind'],
+ app_id=content['app_id'],
+ app_display_name=content['app_display_name'],
+ device_display_name=content['device_display_name'],
+ pushkey=content['pushkey'],
+ lang=content['lang'],
+ 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, _):
+ 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 e9add8e2b4..8e739716ad 100644
--- a/synapse/server.py
+++ b/synapse/server.py
@@ -34,6 +34,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
from synapse.events.builder import EventBuilderFactory
@@ -80,6 +81,7 @@ class BaseHomeServer(object):
'event_sources',
'ratelimiter',
'keyring',
+ 'pusherpool',
'event_builder_factory',
]
@@ -230,6 +232,9 @@ class HomeServer(BaseHomeServer):
hostname=self.hostname,
)
+ 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 4beb951b9f..11706676d0 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -29,6 +29,8 @@ from .stream import StreamStore
from .transactions import TransactionStore
from .keys import KeyStore
from .event_federation import EventFederationStore
+from .pusher import PusherStore
+from .push_rule import PushRuleStore
from .media_repository import MediaRepositoryStore
from .state import StateStore
@@ -60,6 +62,8 @@ SCHEMAS = [
"state",
"event_edges",
"event_signatures",
+ "pusher",
+ "push_rules",
"media_repository",
]
@@ -82,6 +86,8 @@ class DataStore(RoomMemberStore, RoomStore,
DirectoryStore, KeyStore, StateStore, SignatureStore,
EventFederationStore,
MediaRepositoryStore,
+ PusherStore,
+ PushRuleStore
):
def __init__(self, hs):
@@ -382,6 +388,41 @@ class DataStore(RoomMemberStore, RoomStore,
defer.returnValue(events)
@defer.inlineCallbacks
+ def get_room_name_and_aliases(self, room_id):
+ del_sql = (
+ "SELECT event_id FROM redactions WHERE redacts = e.event_id "
+ "LIMIT 1"
+ )
+
+ sql = (
+ "SELECT e.*, (%(redacted)s) AS redacted FROM events as e "
+ "INNER JOIN current_state_events as c ON e.event_id = c.event_id "
+ "INNER JOIN state_events as s ON e.event_id = s.event_id "
+ "WHERE c.room_id = ? "
+ ) % {
+ "redacted": del_sql,
+ }
+
+ sql += " AND (s.type = 'm.room.name' AND s.state_key = '')"
+ sql += " OR s.type = 'm.room.aliases'"
+ args = (room_id,)
+
+ results = yield self._execute_and_decode(sql, *args)
+
+ events = yield self._parse_events(results)
+
+ name = None
+ aliases = []
+
+ for e in events:
+ if e.type == 'm.room.name':
+ name = e.content['name']
+ elif e.type == 'm.room.aliases':
+ aliases.extend(e.content['aliases'])
+
+ defer.returnValue((name, aliases))
+
+ @defer.inlineCallbacks
def _get_min_token(self):
row = yield self._execute(
None,
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index f660fc6eaf..4f172d3967 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -193,6 +193,51 @@ class SQLBaseStore(object):
txn.execute(sql, values.values())
return txn.lastrowid
+ def _simple_upsert(self, table, keyvalues, values):
+ """
+ :param table: The table to upsert into
+ :param keyvalues: Dict of the unique key tables and their new values
+ :param values: Dict of all the nonunique columns and their new values
+ :return: A deferred
+ """
+ return self.runInteraction(
+ "_simple_upsert",
+ self._simple_upsert_txn, table, keyvalues, values
+ )
+
+ def _simple_upsert_txn(self, txn, table, keyvalues, values):
+ # Try to update
+ sql = "UPDATE %s SET %s WHERE %s" % (
+ table,
+ ", ".join("%s = ?" % (k) for k in values),
+ " AND ".join("%s = ?" % (k) for k in keyvalues)
+ )
+ sqlargs = values.values() + keyvalues.values()
+ logger.debug(
+ "[SQL] %s Args=%s",
+ sql, sqlargs,
+ )
+
+ txn.execute(sql, sqlargs)
+ if txn.rowcount == 0:
+ # We didn't update and rows so insert a new one
+ allvalues = {}
+ allvalues.update(keyvalues)
+ allvalues.update(values)
+
+ sql = "INSERT INTO %s (%s) VALUES (%s)" % (
+ table,
+ ", ".join(k for k in allvalues),
+ ", ".join("?" for _ in allvalues)
+ )
+ logger.debug(
+ "[SQL] %s Args=%s",
+ sql, keyvalues.values(),
+ )
+ txn.execute(sql, allvalues.values())
+
+
+
def _simple_select_one(self, table, keyvalues, retcols,
allow_none=False):
"""Executes a SELECT query on the named table, which is expected to
diff --git a/synapse/storage/push_rule.py b/synapse/storage/push_rule.py
new file mode 100644
index 0000000000..76c4557600
--- /dev/null
+++ b/synapse/storage/push_rule.py
@@ -0,0 +1,196 @@
+# -*- 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
+
+import logging
+import copy
+import json
+
+logger = logging.getLogger(__name__)
+
+
+class PushRuleStore(SQLBaseStore):
+ @defer.inlineCallbacks
+ def get_push_rules_for_user_name(self, user_name):
+ sql = (
+ "SELECT "+",".join(PushRuleTable.fields)+
+ "FROM pushers "
+ "WHERE user_name = ?"
+ )
+
+ rows = yield self._execute(None, sql, user_name)
+
+ dicts = []
+ for r in rows:
+ d = {}
+ for i, f in enumerate(PushRuleTable.fields):
+ d[f] = r[i]
+ dicts.append(d)
+
+ defer.returnValue(dicts)
+
+ @defer.inlineCallbacks
+ def add_push_rule(self, **kwargs):
+ vals = copy.copy(kwargs)
+ if 'conditions' in vals:
+ vals['conditions'] = json.dumps(vals['conditions'])
+ if 'actions' in vals:
+ vals['actions'] = json.dumps(vals['actions'])
+ # we could check the rest of the keys are valid column names
+ # but sqlite will do that anyway so I think it's just pointless.
+ if 'id' in vals:
+ del vals['id']
+
+ if 'after' in kwargs or 'before' in kwargs:
+ ret = yield self.runInteraction(
+ "_add_push_rule_relative_txn",
+ self._add_push_rule_relative_txn,
+ **vals
+ )
+ defer.returnValue(ret)
+ else:
+ ret = yield self.runInteraction(
+ "_add_push_rule_highest_priority_txn",
+ self._add_push_rule_highest_priority_txn,
+ **vals
+ )
+ defer.returnValue(ret)
+
+ def _add_push_rule_relative_txn(self, txn, user_name, **kwargs):
+ after = None
+ relative_to_rule = None
+ if 'after' in kwargs and kwargs['after']:
+ after = kwargs['after']
+ relative_to_rule = after
+ if 'before' in kwargs and kwargs['before']:
+ relative_to_rule = kwargs['before']
+
+ # get the priority of the rule we're inserting after/before
+ sql = (
+ "SELECT priority_class, priority FROM "+PushRuleTable.table_name+
+ " WHERE user_name = ? and rule_id = ?"
+ )
+ txn.execute(sql, (user_name, relative_to_rule))
+ res = txn.fetchall()
+ if not res:
+ raise RuleNotFoundException()
+ (priority_class, base_rule_priority) = res[0]
+
+ if 'priority_class' in kwargs and kwargs['priority_class'] != priority_class:
+ raise InconsistentRuleException(
+ "Given priority class does not match class of relative rule"
+ )
+
+ new_rule = copy.copy(kwargs)
+ if 'before' in new_rule:
+ del new_rule['before']
+ if 'after' in new_rule:
+ del new_rule['after']
+ new_rule['priority_class'] = priority_class
+ new_rule['user_name'] = user_name
+
+ # check if the priority before/after is free
+ new_rule_priority = base_rule_priority
+ if after:
+ new_rule_priority -= 1
+ else:
+ new_rule_priority += 1
+
+ new_rule['priority'] = new_rule_priority
+
+ sql = (
+ "SELECT COUNT(*) FROM "+PushRuleTable.table_name+
+ " WHERE user_name = ? AND priority_class = ? AND priority = ?"
+ )
+ txn.execute(sql, (user_name, priority_class, new_rule_priority))
+ res = txn.fetchall()
+ num_conflicting = res[0][0]
+
+ # if there are conflicting rules, bump everything
+ if num_conflicting:
+ sql = "UPDATE "+PushRuleTable.table_name+" SET priority = priority "
+ if after:
+ sql += "-1"
+ else:
+ sql += "+1"
+ sql += " WHERE user_name = ? AND priority_class = ? AND priority "
+ if after:
+ sql += "<= ?"
+ else:
+ sql += ">= ?"
+
+ txn.execute(sql, (user_name, priority_class, new_rule_priority))
+
+ # now insert the new rule
+ sql = "INSERT OR REPLACE INTO "+PushRuleTable.table_name+" ("
+ sql += ",".join(new_rule.keys())+") VALUES ("
+ sql += ", ".join(["?" for _ in new_rule.keys()])+")"
+
+ txn.execute(sql, new_rule.values())
+
+ def _add_push_rule_highest_priority_txn(self, txn, user_name, priority_class, **kwargs):
+ # find the highest priority rule in that class
+ sql = (
+ "SELECT COUNT(*), MAX(priority) FROM "+PushRuleTable.table_name+
+ " WHERE user_name = ? and priority_class = ?"
+ )
+ txn.execute(sql, (user_name, priority_class))
+ res = txn.fetchall()
+ (how_many, highest_prio) = res[0]
+
+ new_prio = 0
+ if how_many > 0:
+ new_prio = highest_prio + 1
+
+ # and insert the new rule
+ new_rule = copy.copy(kwargs)
+ if 'id' in new_rule:
+ del new_rule['id']
+ new_rule['user_name'] = user_name
+ new_rule['priority_class'] = priority_class
+ new_rule['priority'] = new_prio
+
+ sql = "INSERT OR REPLACE INTO "+PushRuleTable.table_name+" ("
+ sql += ",".join(new_rule.keys())+") VALUES ("
+ sql += ", ".join(["?" for _ in new_rule.keys()])+")"
+
+ txn.execute(sql, new_rule.values())
+
+class RuleNotFoundException(Exception):
+ pass
+
+
+class InconsistentRuleException(Exception):
+ pass
+
+
+class PushRuleTable(Table):
+ table_name = "push_rules"
+
+ fields = [
+ "id",
+ "user_name",
+ "rule_id",
+ "priority_class",
+ "priority",
+ "conditions",
+ "actions",
+ ]
+
+ EntryType = collections.namedtuple("PushRuleEntry", fields)
\ No newline at end of file
diff --git a/synapse/storage/pusher.py b/synapse/storage/pusher.py
new file mode 100644
index 0000000000..113cdc8a8e
--- /dev/null
+++ b/synapse/storage/pusher.py
@@ -0,0 +1,173 @@
+# -*- 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 synapse.api.errors import StoreError
+
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class PusherStore(SQLBaseStore):
+ @defer.inlineCallbacks
+ def get_pushers_by_app_id_and_pushkey(self, app_id_and_pushkey):
+ sql = (
+ "SELECT id, user_name, kind, instance_handle, app_id,"
+ "app_display_name, device_display_name, pushkey, ts, data, "
+ "last_token, last_success, failing_since "
+ "FROM pushers "
+ "WHERE app_id = ? AND pushkey = ?"
+ )
+
+ rows = yield self._execute(
+ None, sql, app_id_and_pushkey[0], app_id_and_pushkey[1]
+ )
+
+ ret = [
+ {
+ "id": r[0],
+ "user_name": r[1],
+ "kind": r[2],
+ "instance_handle": r[3],
+ "app_id": r[4],
+ "app_display_name": r[5],
+ "device_display_name": r[6],
+ "pushkey": r[7],
+ "pushkey_ts": r[8],
+ "data": r[9],
+ "last_token": r[10],
+ "last_success": r[11],
+ "failing_since": r[12]
+ }
+ for r in rows
+ ]
+
+ defer.returnValue(ret[0])
+
+ @defer.inlineCallbacks
+ def get_all_pushers(self):
+ sql = (
+ "SELECT id, user_name, kind, instance_handle, app_id,"
+ "app_display_name, device_display_name, pushkey, ts, data, "
+ "last_token, last_success, failing_since "
+ "FROM pushers"
+ )
+
+ rows = yield self._execute(None, sql)
+
+ ret = [
+ {
+ "id": r[0],
+ "user_name": r[1],
+ "kind": r[2],
+ "instance_handle": r[3],
+ "app_id": r[4],
+ "app_display_name": r[5],
+ "device_display_name": r[6],
+ "pushkey": r[7],
+ "pushkey_ts": r[8],
+ "data": r[9],
+ "last_token": r[10],
+ "last_success": r[11],
+ "failing_since": r[12]
+ }
+ for r in rows
+ ]
+
+ defer.returnValue(ret)
+
+ @defer.inlineCallbacks
+ def add_pusher(self, user_name, instance_handle, kind, app_id,
+ app_display_name, device_display_name,
+ pushkey, pushkey_ts, lang, data):
+ try:
+ yield self._simple_upsert(
+ PushersTable.table_name,
+ dict(
+ app_id=app_id,
+ pushkey=pushkey,
+ ),
+ dict(
+ user_name=user_name,
+ kind=kind,
+ instance_handle=instance_handle,
+ app_display_name=app_display_name,
+ device_display_name=device_display_name,
+ ts=pushkey_ts,
+ lang=lang,
+ data=data
+ ))
+ except Exception as e:
+ logger.error("create_pusher with failed: %s", e)
+ raise StoreError(500, "Problem creating pusher.")
+
+ @defer.inlineCallbacks
+ def delete_pusher_by_app_id_pushkey(self, app_id, pushkey):
+ yield self._simple_delete_one(
+ PushersTable.table_name,
+ dict(app_id=app_id, pushkey=pushkey)
+ )
+
+ @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}
+ )
+
+ @defer.inlineCallbacks
+ def update_pusher_last_token_and_success(self, user_name, pushkey,
+ last_token, last_success):
+ yield self._simple_update_one(
+ PushersTable.table_name,
+ {'user_name': user_name, 'pushkey': pushkey},
+ {'last_token': last_token, 'last_success': last_success}
+ )
+
+ @defer.inlineCallbacks
+ def update_pusher_failing_since(self, user_name, pushkey, failing_since):
+ yield self._simple_update_one(
+ PushersTable.table_name,
+ {'user_name': user_name, 'pushkey': pushkey},
+ {'failing_since': failing_since}
+ )
+
+
+class PushersTable(Table):
+ table_name = "pushers"
+
+ fields = [
+ "id",
+ "user_name",
+ "kind",
+ "instance_handle",
+ "app_id",
+ "app_display_name",
+ "device_display_name",
+ "pushkey",
+ "pushkey_ts",
+ "data",
+ "last_token",
+ "last_success",
+ "failing_since"
+ ]
+
+ EntryType = collections.namedtuple("PusherEntry", fields)
\ No newline at end of file
diff --git a/synapse/storage/schema/delta/v10.sql b/synapse/storage/schema/delta/v10.sql
new file mode 100644
index 0000000000..8c4dfd5c1b
--- /dev/null
+++ b/synapse/storage/schema/delta/v10.sql
@@ -0,0 +1,46 @@
+/* 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,
+ instance_handle varchar(32) NOT NULL,
+ kind varchar(8) NOT NULL,
+ app_id varchar(64) NOT NULL,
+ app_display_name varchar(64) NOT NULL,
+ device_display_name varchar(128) NOT NULL,
+ pushkey blob NOT NULL,
+ ts BIGINT NOT NULL,
+ lang varchar(8),
+ data blob,
+ last_token TEXT,
+ last_success BIGINT,
+ failing_since BIGINT,
+ FOREIGN KEY(user_name) REFERENCES users(name),
+ UNIQUE (app_id, pushkey)
+);
+
+CREATE TABLE IF NOT EXISTS push_rules (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_name TEXT NOT NULL,
+ rule_id TEXT NOT NULL,
+ priority_class TINYINT NOT NULL,
+ priority INTEGER NOT NULL DEFAULT 0,
+ conditions TEXT NOT NULL,
+ actions TEXT NOT NULL,
+ UNIQUE(user_name, rule_id)
+);
+
+CREATE INDEX IF NOT EXISTS push_rules_user_name on push_rules (user_name);
diff --git a/synapse/storage/schema/pusher.sql b/synapse/storage/schema/pusher.sql
new file mode 100644
index 0000000000..8c4dfd5c1b
--- /dev/null
+++ b/synapse/storage/schema/pusher.sql
@@ -0,0 +1,46 @@
+/* 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,
+ instance_handle varchar(32) NOT NULL,
+ kind varchar(8) NOT NULL,
+ app_id varchar(64) NOT NULL,
+ app_display_name varchar(64) NOT NULL,
+ device_display_name varchar(128) NOT NULL,
+ pushkey blob NOT NULL,
+ ts BIGINT NOT NULL,
+ lang varchar(8),
+ data blob,
+ last_token TEXT,
+ last_success BIGINT,
+ failing_since BIGINT,
+ FOREIGN KEY(user_name) REFERENCES users(name),
+ UNIQUE (app_id, pushkey)
+);
+
+CREATE TABLE IF NOT EXISTS push_rules (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ user_name TEXT NOT NULL,
+ rule_id TEXT NOT NULL,
+ priority_class TINYINT NOT NULL,
+ priority INTEGER NOT NULL DEFAULT 0,
+ conditions TEXT NOT NULL,
+ actions TEXT NOT NULL,
+ UNIQUE(user_name, rule_id)
+);
+
+CREATE INDEX IF NOT EXISTS push_rules_user_name on push_rules (user_name);
|
