Matrix-Mirrors/anonymousland-synapse
1
0
Fork 0
mirror of https://git.anonymousland.org/anonymousland/synapse.git synced 2025-12-15 17:08:57 -05:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #41 from matrix-org/client_v2_sync

Browse source 
Client v2 sync
This commit is contained in:
Mark Haines 2015-01-30 15:54:25 +00:00
commit 6dc92d3427
7 changed files with 794 additions and 46 deletions
Download patch file Download diff file Expand all files Collapse all files

60
synapse/storage/stream.py
View file

@ -181,8 +181,11 @@ class StreamStore(SQLBaseStore):
get_prev_content=True
)
self._set_before_and_after(ret, rows)
if rows:
key = "s%d" % max([r["stream_ordering"] for r in rows])
else:
# Assume we didn't get anything because there was nothing to
# get.
@ -260,22 +263,44 @@ class StreamStore(SQLBaseStore):
get_prev_content=True
)
self._set_before_and_after(events, rows)
return events, next_token,
return self.runInteraction("paginate_room_events", f)
def get_recent_events_for_room(self, room_id, limit, end_token,
with_feedback=False):
with_feedback=False, from_token=None):
# TODO (erikj): Handle compressed feedback
sql = (
"SELECT stream_ordering, topological_ordering, event_id FROM events "
"WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0 "
"ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? "
)
end_token = _StreamToken.parse_stream_token(end_token)
def f(txn):
txn.execute(sql, (room_id, end_token, limit,))
if from_token is None:
sql = (
"SELECT stream_ordering, topological_ordering, event_id"
" FROM events"
" WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0"
" ORDER BY topological_ordering DESC, stream_ordering DESC"
" LIMIT ?"
)
else:
from_token = _StreamToken.parse_stream_token(from_token)
sql = (
"SELECT stream_ordering, topological_ordering, event_id"
" FROM events"
" WHERE room_id = ? AND stream_ordering > ?"
" AND stream_ordering <= ? AND outlier = 0"
" ORDER BY topological_ordering DESC, stream_ordering DESC"
" LIMIT ?"
)
def get_recent_events_for_room_txn(txn):
if from_token is None:
txn.execute(sql, (room_id, end_token.stream, limit,))
else:
txn.execute(sql, (
room_id, from_token.stream, end_token.stream, limit
))
rows = self.cursor_to_dict(txn)
@ -291,9 +316,9 @@ class StreamStore(SQLBaseStore):
toke = rows[0]["stream_ordering"] - 1
start_token = str(_StreamToken(topo, toke))
token = (start_token, end_token)
token = (start_token, str(end_token))
else:
token = (end_token, end_token)
token = (str(end_token), str(end_token))
events = self._get_events_txn(
txn,
@ -301,9 +326,13 @@ class StreamStore(SQLBaseStore):
get_prev_content=True
)
self._set_before_and_after(events, rows)
return events, token
return self.runInteraction("get_recent_events_for_room", f)
return self.runInteraction(
"get_recent_events_for_room", get_recent_events_for_room_txn
)
def get_room_events_max_id(self):
return self.runInteraction(
@ -325,3 +354,12 @@ class StreamStore(SQLBaseStore):
key = res[0]["m"]
return "s%d" % (key,)
@staticmethod
def _set_before_and_after(events, rows):
for event, row in zip(events, rows):
stream = row["stream_ordering"]
topo = event.depth
internal = event.internal_metadata
internal.before = str(_StreamToken(topo, stream - 1))
internal.after = str(_StreamToken(topo, stream))