Matrix-Mirrors/synapse-product
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse-product.git synced 2024-10-01 08:25:44 -04:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #549 from matrix-org/erikj/sync

Browse Source 
Fetch events in a separate transaction.
This commit is contained in:
Erik Johnston 2016-02-02 16:23:54 +00:00
commit 43e13dbd4d
3 changed files with 32 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
synapse/handlers/room.py
View File

@ -1013,7 +1013,7 @@ class RoomEventSource(object):
limit=limit, limit=limit,
) )
else: else:
room_events = yield self.store.get_room_changes_for_user( room_events = yield self.store.get_membership_changes_for_user(
user.to_string(), from_key, to_key user.to_string(), from_key, to_key
) )

2
synapse/handlers/sync.py
View File

@ -479,7 +479,7 @@ class SyncHandler(BaseHandler):
) )
# Get a list of membership change events that have happened. # Get a list of membership change events that have happened.
rooms_changed = yield self.store.get_room_changes_for_user( rooms_changed = yield self.store.get_membership_changes_for_user(
user_id, since_token.room_key, now_token.room_key user_id, since_token.room_key, now_token.room_key
) )

57
synapse/storage/stream.py
View File

@ -220,28 +220,30 @@ class StreamStore(SQLBaseStore):
rows = self.cursor_to_dict(txn) rows = self.cursor_to_dict(txn)
ret = self._get_events_txn( return rows
txn,
[r["event_id"] for r in rows],
get_prev_content=True
)
self._set_before_and_after(ret, rows, topo_order=False) rows = yield self.runInteraction("get_room_events_stream_for_room", f)
ret.reverse() ret = yield self._get_events(
[r["event_id"] for r in rows],
get_prev_content=True
)
if rows: self._set_before_and_after(ret, rows, topo_order=False)
key = "s%d" % min(r["stream_ordering"] for r in rows)
else:
# Assume we didn't get anything because there was nothing to
# get.
key = from_key
return ret, key ret.reverse()
res = yield self.runInteraction("get_room_events_stream_for_room", f)
defer.returnValue(res)
def get_room_changes_for_user(self, user_id, from_key, to_key): if rows:
key = "s%d" % min(r["stream_ordering"] for r in rows)
else:
# Assume we didn't get anything because there was nothing to
# get.
key = from_key
defer.returnValue((ret, key))
@defer.inlineCallbacks
def get_membership_changes_for_user(self, user_id, from_key, to_key):
if from_key is not None: if from_key is not None:
from_id = RoomStreamToken.parse_stream_token(from_key).stream from_id = RoomStreamToken.parse_stream_token(from_key).stream
else: else:
@ -249,14 +251,14 @@ class StreamStore(SQLBaseStore):
to_id = RoomStreamToken.parse_stream_token(to_key).stream to_id = RoomStreamToken.parse_stream_token(to_key).stream
if from_key == to_key: if from_key == to_key:
return defer.succeed([]) defer.returnValue([])
if from_id: if from_id:
has_changed = self._membership_stream_cache.has_entity_changed( has_changed = self._membership_stream_cache.has_entity_changed(
user_id, int(from_id) user_id, int(from_id)
) )
if not has_changed: if not has_changed:
return defer.succeed([]) defer.returnValue([])
def f(txn): def f(txn):
if from_id is not None: if from_id is not None:
@ -281,17 +283,18 @@ class StreamStore(SQLBaseStore):
txn.execute(sql, (user_id, to_id,)) txn.execute(sql, (user_id, to_id,))
rows = self.cursor_to_dict(txn) rows = self.cursor_to_dict(txn)
ret = self._get_events_txn( return rows
txn,
[r["event_id"] for r in rows],
get_prev_content=True
)
self._set_before_and_after(ret, rows, topo_order=False) rows = yield self.runInteraction("get_membership_changes_for_user", f)
return ret ret = yield self._get_events(
[r["event_id"] for r in rows],
get_prev_content=True
)
return self.runInteraction("get_room_changes_for_user", f) self._set_before_and_after(ret, rows, topo_order=False)
defer.returnValue(ret)
def get_room_events_stream( def get_room_events_stream(
self, self,