mirror of
https://mau.dev/maunium/synapse.git
synced 2024-12-14 17:23:53 +01:00
An initial hack at storing presence state-change mtimes in database and presenting age durations to clients/federation events
This commit is contained in:
parent
a6a9b71da0
commit
d05aa651f8
7 changed files with 101 additions and 29 deletions
|
@ -56,6 +56,8 @@ class PresenceHandler(BaseHandler):
|
||||||
|
|
||||||
self.homeserver = hs
|
self.homeserver = hs
|
||||||
|
|
||||||
|
self.clock = hs.get_clock()
|
||||||
|
|
||||||
distributor = hs.get_distributor()
|
distributor = hs.get_distributor()
|
||||||
distributor.observe("registered_user", self.registered_user)
|
distributor.observe("registered_user", self.registered_user)
|
||||||
|
|
||||||
|
@ -168,14 +170,15 @@ class PresenceHandler(BaseHandler):
|
||||||
state = yield self.store.get_presence_state(
|
state = yield self.store.get_presence_state(
|
||||||
target_user.localpart
|
target_user.localpart
|
||||||
)
|
)
|
||||||
defer.returnValue(state)
|
|
||||||
else:
|
else:
|
||||||
raise SynapseError(404, "Presence information not visible")
|
raise SynapseError(404, "Presence information not visible")
|
||||||
else:
|
else:
|
||||||
# TODO(paul): Have remote server send us permissions set
|
# TODO(paul): Have remote server send us permissions set
|
||||||
defer.returnValue(
|
state = self._get_or_offline_usercache(target_user).get_state()
|
||||||
self._get_or_offline_usercache(target_user).get_state()
|
|
||||||
)
|
if "mtime" in state:
|
||||||
|
state["mtime_age"] = self.clock.time_msec() - state.pop("mtime")
|
||||||
|
defer.returnValue(state)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def set_state(self, target_user, auth_user, state):
|
def set_state(self, target_user, auth_user, state):
|
||||||
|
@ -209,6 +212,8 @@ class PresenceHandler(BaseHandler):
|
||||||
),
|
),
|
||||||
])
|
])
|
||||||
|
|
||||||
|
state["mtime"] = self.clock.time_msec()
|
||||||
|
|
||||||
now_online = state["state"] != PresenceState.OFFLINE
|
now_online = state["state"] != PresenceState.OFFLINE
|
||||||
was_polling = target_user in self._user_cachemap
|
was_polling = target_user in self._user_cachemap
|
||||||
|
|
||||||
|
@ -361,6 +366,8 @@ class PresenceHandler(BaseHandler):
|
||||||
observed_user = self.hs.parse_userid(p.pop("observed_user_id"))
|
observed_user = self.hs.parse_userid(p.pop("observed_user_id"))
|
||||||
p["observed_user"] = observed_user
|
p["observed_user"] = observed_user
|
||||||
p.update(self._get_or_offline_usercache(observed_user).get_state())
|
p.update(self._get_or_offline_usercache(observed_user).get_state())
|
||||||
|
if "mtime" in p:
|
||||||
|
p["mtime_age"] = self.clock.time_msec() - p.pop("mtime")
|
||||||
|
|
||||||
defer.returnValue(presence)
|
defer.returnValue(presence)
|
||||||
|
|
||||||
|
@ -546,10 +553,15 @@ class PresenceHandler(BaseHandler):
|
||||||
def _push_presence_remote(self, user, destination, state=None):
|
def _push_presence_remote(self, user, destination, state=None):
|
||||||
if state is None:
|
if state is None:
|
||||||
state = yield self.store.get_presence_state(user.localpart)
|
state = yield self.store.get_presence_state(user.localpart)
|
||||||
|
|
||||||
yield self.distributor.fire(
|
yield self.distributor.fire(
|
||||||
"collect_presencelike_data", user, state
|
"collect_presencelike_data", user, state
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if "mtime" in state:
|
||||||
|
state = dict(state)
|
||||||
|
state["mtime_age"] = self.clock.time_msec() - state.pop("mtime")
|
||||||
|
|
||||||
yield self.federation.send_edu(
|
yield self.federation.send_edu(
|
||||||
destination=destination,
|
destination=destination,
|
||||||
edu_type="m.presence",
|
edu_type="m.presence",
|
||||||
|
@ -585,6 +597,9 @@ class PresenceHandler(BaseHandler):
|
||||||
state = dict(push)
|
state = dict(push)
|
||||||
del state["user_id"]
|
del state["user_id"]
|
||||||
|
|
||||||
|
if "mtime_age" in state:
|
||||||
|
state["mtime"] = self.clock.time_msec() - state.pop("mtime_age")
|
||||||
|
|
||||||
statuscache = self._get_or_make_usercache(user)
|
statuscache = self._get_or_make_usercache(user)
|
||||||
|
|
||||||
self._user_cachemap_latest_serial += 1
|
self._user_cachemap_latest_serial += 1
|
||||||
|
@ -631,9 +646,14 @@ class PresenceHandler(BaseHandler):
|
||||||
|
|
||||||
def push_update_to_clients(self, observer_user, observed_user,
|
def push_update_to_clients(self, observer_user, observed_user,
|
||||||
statuscache):
|
statuscache):
|
||||||
|
state = statuscache.make_event(user=observed_user, clock=self.clock)
|
||||||
|
|
||||||
self.notifier.on_new_user_event(
|
self.notifier.on_new_user_event(
|
||||||
observer_user.to_string(),
|
observer_user.to_string(),
|
||||||
event_data=statuscache.make_event(user=observed_user),
|
event_data=statuscache.make_event(
|
||||||
|
user=observed_user,
|
||||||
|
clock=self.clock
|
||||||
|
),
|
||||||
stream_type=PresenceStreamData,
|
stream_type=PresenceStreamData,
|
||||||
store_id=statuscache.serial
|
store_id=statuscache.serial
|
||||||
)
|
)
|
||||||
|
@ -652,8 +672,10 @@ class PresenceStreamData(StreamData):
|
||||||
if from_key < cachemap[k].serial <= to_key]
|
if from_key < cachemap[k].serial <= to_key]
|
||||||
|
|
||||||
if updates:
|
if updates:
|
||||||
|
clock = self.presence.clock
|
||||||
|
|
||||||
latest_serial = max([x[1].serial for x in updates])
|
latest_serial = max([x[1].serial for x in updates])
|
||||||
data = [x[1].make_event(user=x[0]) for x in updates]
|
data = [x[1].make_event(user=x[0], clock=clock) for x in updates]
|
||||||
return ((data, latest_serial))
|
return ((data, latest_serial))
|
||||||
else:
|
else:
|
||||||
return (([], self.presence._user_cachemap_latest_serial))
|
return (([], self.presence._user_cachemap_latest_serial))
|
||||||
|
@ -674,6 +696,8 @@ class UserPresenceCache(object):
|
||||||
self.serial = None
|
self.serial = None
|
||||||
|
|
||||||
def update(self, state, serial):
|
def update(self, state, serial):
|
||||||
|
assert("mtime_age" not in state)
|
||||||
|
|
||||||
self.state.update(state)
|
self.state.update(state)
|
||||||
# Delete keys that are now 'None'
|
# Delete keys that are now 'None'
|
||||||
for k in self.state.keys():
|
for k in self.state.keys():
|
||||||
|
@ -691,8 +715,11 @@ class UserPresenceCache(object):
|
||||||
# clone it so caller can't break our cache
|
# clone it so caller can't break our cache
|
||||||
return dict(self.state)
|
return dict(self.state)
|
||||||
|
|
||||||
def make_event(self, user):
|
def make_event(self, user, clock):
|
||||||
content = self.get_state()
|
content = self.get_state()
|
||||||
content["user_id"] = user.to_string()
|
content["user_id"] = user.to_string()
|
||||||
|
|
||||||
|
if "mtime" in content:
|
||||||
|
content["mtime_age"] = clock.time_msec() - content.pop("mtime")
|
||||||
|
|
||||||
return {"type": "m.presence", "content": content}
|
return {"type": "m.presence", "content": content}
|
||||||
|
|
|
@ -29,6 +29,7 @@ class SQLBaseStore(object):
|
||||||
def __init__(self, hs):
|
def __init__(self, hs):
|
||||||
self.hs = hs
|
self.hs = hs
|
||||||
self._db_pool = hs.get_db_pool()
|
self._db_pool = hs.get_db_pool()
|
||||||
|
self._clock = hs.get_clock()
|
||||||
|
|
||||||
def cursor_to_dict(self, cursor):
|
def cursor_to_dict(self, cursor):
|
||||||
"""Converts a SQL cursor into an list of dicts.
|
"""Converts a SQL cursor into an list of dicts.
|
||||||
|
|
|
@ -35,7 +35,7 @@ class PresenceStore(SQLBaseStore):
|
||||||
return self._simple_select_one(
|
return self._simple_select_one(
|
||||||
table="presence",
|
table="presence",
|
||||||
keyvalues={"user_id": user_localpart},
|
keyvalues={"user_id": user_localpart},
|
||||||
retcols=["state", "status_msg"],
|
retcols=["state", "status_msg", "mtime"],
|
||||||
)
|
)
|
||||||
|
|
||||||
def set_presence_state(self, user_localpart, new_state):
|
def set_presence_state(self, user_localpart, new_state):
|
||||||
|
@ -43,7 +43,8 @@ class PresenceStore(SQLBaseStore):
|
||||||
table="presence",
|
table="presence",
|
||||||
keyvalues={"user_id": user_localpart},
|
keyvalues={"user_id": user_localpart},
|
||||||
updatevalues={"state": new_state["state"],
|
updatevalues={"state": new_state["state"],
|
||||||
"status_msg": new_state["status_msg"]},
|
"status_msg": new_state["status_msg"],
|
||||||
|
"mtime": self._clock.time_msec()},
|
||||||
retcols=["state"],
|
retcols=["state"],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
@ -16,6 +16,7 @@ CREATE TABLE IF NOT EXISTS presence(
|
||||||
user_id INTEGER NOT NULL,
|
user_id INTEGER NOT NULL,
|
||||||
state INTEGER,
|
state INTEGER,
|
||||||
status_msg TEXT,
|
status_msg TEXT,
|
||||||
|
mtime INTEGER, -- miliseconds since last state change
|
||||||
FOREIGN KEY(user_id) REFERENCES users(id)
|
FOREIGN KEY(user_id) REFERENCES users(id)
|
||||||
);
|
);
|
||||||
|
|
||||||
|
|
|
@ -20,6 +20,8 @@ from twisted.internet import defer
|
||||||
from mock import Mock, call, ANY
|
from mock import Mock, call, ANY
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from ..utils import MockClock
|
||||||
|
|
||||||
from synapse.server import HomeServer
|
from synapse.server import HomeServer
|
||||||
from synapse.api.constants import PresenceState
|
from synapse.api.constants import PresenceState
|
||||||
from synapse.api.errors import SynapseError
|
from synapse.api.errors import SynapseError
|
||||||
|
@ -55,6 +57,7 @@ class PresenceStateTestCase(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
hs = HomeServer("test",
|
hs = HomeServer("test",
|
||||||
|
clock=MockClock(),
|
||||||
db_pool=None,
|
db_pool=None,
|
||||||
datastore=Mock(spec=[
|
datastore=Mock(spec=[
|
||||||
"get_presence_state",
|
"get_presence_state",
|
||||||
|
@ -154,7 +157,11 @@ class PresenceStateTestCase(unittest.TestCase):
|
||||||
mocked_set.assert_called_with("apple",
|
mocked_set.assert_called_with("apple",
|
||||||
{"state": UNAVAILABLE, "status_msg": "Away"})
|
{"state": UNAVAILABLE, "status_msg": "Away"})
|
||||||
self.mock_start.assert_called_with(self.u_apple,
|
self.mock_start.assert_called_with(self.u_apple,
|
||||||
state={"state": UNAVAILABLE, "status_msg": "Away"})
|
state={
|
||||||
|
"state": UNAVAILABLE,
|
||||||
|
"status_msg": "Away",
|
||||||
|
"mtime": 1000000, # MockClock
|
||||||
|
})
|
||||||
|
|
||||||
yield self.handler.set_state(
|
yield self.handler.set_state(
|
||||||
target_user=self.u_apple, auth_user=self.u_apple,
|
target_user=self.u_apple, auth_user=self.u_apple,
|
||||||
|
@ -386,7 +393,10 @@ class PresencePushTestCase(unittest.TestCase):
|
||||||
self.replication.send_edu = Mock()
|
self.replication.send_edu = Mock()
|
||||||
self.replication.send_edu.return_value = defer.succeed((200, "OK"))
|
self.replication.send_edu.return_value = defer.succeed((200, "OK"))
|
||||||
|
|
||||||
|
self.clock = MockClock()
|
||||||
|
|
||||||
hs = HomeServer("test",
|
hs = HomeServer("test",
|
||||||
|
clock=self.clock,
|
||||||
db_pool=None,
|
db_pool=None,
|
||||||
datastore=Mock(spec=[
|
datastore=Mock(spec=[
|
||||||
"set_presence_state",
|
"set_presence_state",
|
||||||
|
@ -519,13 +529,18 @@ class PresencePushTestCase(unittest.TestCase):
|
||||||
yield self.handler.set_state(self.u_banana, self.u_banana,
|
yield self.handler.set_state(self.u_banana, self.u_banana,
|
||||||
{"state": ONLINE})
|
{"state": ONLINE})
|
||||||
|
|
||||||
|
self.clock.advance_time(2)
|
||||||
|
|
||||||
presence = yield self.handler.get_presence_list(
|
presence = yield self.handler.get_presence_list(
|
||||||
observer_user=self.u_apple, accepted=True)
|
observer_user=self.u_apple, accepted=True)
|
||||||
|
|
||||||
self.assertEquals([
|
self.assertEquals([
|
||||||
{"observed_user": self.u_banana, "state": ONLINE},
|
{"observed_user": self.u_banana,
|
||||||
{"observed_user": self.u_clementine, "state": OFFLINE}],
|
"state": ONLINE,
|
||||||
presence)
|
"mtime_age": 2000},
|
||||||
|
{"observed_user": self.u_clementine,
|
||||||
|
"state": OFFLINE},
|
||||||
|
], presence)
|
||||||
|
|
||||||
self.mock_update_client.assert_has_calls([
|
self.mock_update_client.assert_has_calls([
|
||||||
call(observer_user=self.u_banana,
|
call(observer_user=self.u_banana,
|
||||||
|
@ -555,7 +570,8 @@ class PresencePushTestCase(unittest.TestCase):
|
||||||
content={
|
content={
|
||||||
"push": [
|
"push": [
|
||||||
{"user_id": "@apple:test",
|
{"user_id": "@apple:test",
|
||||||
"state": "online"},
|
"state": "online",
|
||||||
|
"mtime_age": 0},
|
||||||
],
|
],
|
||||||
}),
|
}),
|
||||||
call(
|
call(
|
||||||
|
@ -564,7 +580,8 @@ class PresencePushTestCase(unittest.TestCase):
|
||||||
content={
|
content={
|
||||||
"push": [
|
"push": [
|
||||||
{"user_id": "@apple:test",
|
{"user_id": "@apple:test",
|
||||||
"state": "online"},
|
"state": "online",
|
||||||
|
"mtime_age": 0},
|
||||||
],
|
],
|
||||||
})
|
})
|
||||||
], any_order=True)
|
], any_order=True)
|
||||||
|
@ -582,7 +599,8 @@ class PresencePushTestCase(unittest.TestCase):
|
||||||
"remote", "m.presence", {
|
"remote", "m.presence", {
|
||||||
"push": [
|
"push": [
|
||||||
{"user_id": "@potato:remote",
|
{"user_id": "@potato:remote",
|
||||||
"state": "online"},
|
"state": "online",
|
||||||
|
"mtime_age": 1000},
|
||||||
],
|
],
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
@ -596,9 +614,11 @@ class PresencePushTestCase(unittest.TestCase):
|
||||||
statuscache=ANY),
|
statuscache=ANY),
|
||||||
], any_order=True)
|
], any_order=True)
|
||||||
|
|
||||||
|
self.clock.advance_time(2)
|
||||||
|
|
||||||
state = yield self.handler.get_state(self.u_potato, self.u_apple)
|
state = yield self.handler.get_state(self.u_potato, self.u_apple)
|
||||||
|
|
||||||
self.assertEquals({"state": ONLINE}, state)
|
self.assertEquals({"state": ONLINE, "mtime_age": 3000}, state)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def test_join_room_local(self):
|
def test_join_room_local(self):
|
||||||
|
|
|
@ -22,6 +22,8 @@ from twisted.internet import defer
|
||||||
from mock import Mock, call, ANY
|
from mock import Mock, call, ANY
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from ..utils import MockClock
|
||||||
|
|
||||||
from synapse.server import HomeServer
|
from synapse.server import HomeServer
|
||||||
from synapse.api.constants import PresenceState
|
from synapse.api.constants import PresenceState
|
||||||
from synapse.handlers.presence import PresenceHandler
|
from synapse.handlers.presence import PresenceHandler
|
||||||
|
@ -60,6 +62,7 @@ class PresenceProfilelikeDataTestCase(unittest.TestCase):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
hs = HomeServer("test",
|
hs = HomeServer("test",
|
||||||
|
clock=MockClock(),
|
||||||
db_pool=None,
|
db_pool=None,
|
||||||
datastore=Mock(spec=[
|
datastore=Mock(spec=[
|
||||||
"set_presence_state",
|
"set_presence_state",
|
||||||
|
@ -156,10 +159,14 @@ class PresenceProfilelikeDataTestCase(unittest.TestCase):
|
||||||
observer_user=self.u_apple, accepted=True)
|
observer_user=self.u_apple, accepted=True)
|
||||||
|
|
||||||
self.assertEquals([
|
self.assertEquals([
|
||||||
{"observed_user": self.u_banana, "state": ONLINE,
|
{"observed_user": self.u_banana,
|
||||||
"displayname": "Frank", "avatar_url": "http://foo"},
|
"state": ONLINE,
|
||||||
{"observed_user": self.u_clementine, "state": OFFLINE}],
|
"mtime_age": 0,
|
||||||
presence)
|
"displayname": "Frank",
|
||||||
|
"avatar_url": "http://foo"},
|
||||||
|
{"observed_user": self.u_clementine,
|
||||||
|
"state": OFFLINE}],
|
||||||
|
presence)
|
||||||
|
|
||||||
self.mock_update_client.assert_has_calls([
|
self.mock_update_client.assert_has_calls([
|
||||||
call(observer_user=self.u_apple,
|
call(observer_user=self.u_apple,
|
||||||
|
@ -171,9 +178,12 @@ class PresenceProfilelikeDataTestCase(unittest.TestCase):
|
||||||
], any_order=True)
|
], any_order=True)
|
||||||
|
|
||||||
statuscache = self.mock_update_client.call_args[1]["statuscache"]
|
statuscache = self.mock_update_client.call_args[1]["statuscache"]
|
||||||
self.assertEquals({"state": ONLINE,
|
self.assertEquals({
|
||||||
"displayname": "Frank",
|
"state": ONLINE,
|
||||||
"avatar_url": "http://foo"}, statuscache.state)
|
"mtime": 1000000, # MockClock
|
||||||
|
"displayname": "Frank",
|
||||||
|
"avatar_url": "http://foo",
|
||||||
|
}, statuscache.state)
|
||||||
|
|
||||||
self.mock_update_client.reset_mock()
|
self.mock_update_client.reset_mock()
|
||||||
|
|
||||||
|
@ -193,9 +203,12 @@ class PresenceProfilelikeDataTestCase(unittest.TestCase):
|
||||||
], any_order=True)
|
], any_order=True)
|
||||||
|
|
||||||
statuscache = self.mock_update_client.call_args[1]["statuscache"]
|
statuscache = self.mock_update_client.call_args[1]["statuscache"]
|
||||||
self.assertEquals({"state": ONLINE,
|
self.assertEquals({
|
||||||
"displayname": "I am an Apple",
|
"state": ONLINE,
|
||||||
"avatar_url": "http://foo"}, statuscache.state)
|
"mtime": 1000000, # MockClock
|
||||||
|
"displayname": "I am an Apple",
|
||||||
|
"avatar_url": "http://foo",
|
||||||
|
}, statuscache.state)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
def test_push_remote(self):
|
def test_push_remote(self):
|
||||||
|
@ -220,6 +233,7 @@ class PresenceProfilelikeDataTestCase(unittest.TestCase):
|
||||||
"push": [
|
"push": [
|
||||||
{"user_id": "@apple:test",
|
{"user_id": "@apple:test",
|
||||||
"state": "online",
|
"state": "online",
|
||||||
|
"mtime_age": 0,
|
||||||
"displayname": "Frank",
|
"displayname": "Frank",
|
||||||
"avatar_url": "http://foo"},
|
"avatar_url": "http://foo"},
|
||||||
],
|
],
|
||||||
|
|
|
@ -234,7 +234,11 @@ class PresenceEventStreamTestCase(unittest.TestCase):
|
||||||
# I'll already get my own presence state change
|
# I'll already get my own presence state change
|
||||||
self.assertEquals({"start": "0", "end": "1", "chunk": [
|
self.assertEquals({"start": "0", "end": "1", "chunk": [
|
||||||
{"type": "m.presence",
|
{"type": "m.presence",
|
||||||
"content": {"user_id": "@apple:test", "state": ONLINE}},
|
"content": {
|
||||||
|
"user_id": "@apple:test",
|
||||||
|
"state": ONLINE,
|
||||||
|
"mtime_age": 0,
|
||||||
|
}},
|
||||||
]}, response)
|
]}, response)
|
||||||
|
|
||||||
self.mock_datastore.set_presence_state.return_value = defer.succeed(
|
self.mock_datastore.set_presence_state.return_value = defer.succeed(
|
||||||
|
@ -251,5 +255,9 @@ class PresenceEventStreamTestCase(unittest.TestCase):
|
||||||
self.assertEquals(200, code)
|
self.assertEquals(200, code)
|
||||||
self.assertEquals({"start": "1", "end": "2", "chunk": [
|
self.assertEquals({"start": "1", "end": "2", "chunk": [
|
||||||
{"type": "m.presence",
|
{"type": "m.presence",
|
||||||
"content": {"user_id": "@banana:test", "state": ONLINE}},
|
"content": {
|
||||||
|
"user_id": "@banana:test",
|
||||||
|
"state": ONLINE,
|
||||||
|
"mtime_age": 0,
|
||||||
|
}},
|
||||||
]}, response)
|
]}, response)
|
||||||
|
|
Loading…
Reference in a new issue