diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py
index c96ce6f1d3..59719a1fae 100644
--- a/synapse/handlers/room.py
+++ b/synapse/handlers/room.py
@@ -245,14 +245,12 @@ class RoomMemberHandler(BaseHandler):
self.distributor.declare("user_left_room")
@defer.inlineCallbacks
- def get_room_members(self, room_id, membership=Membership.JOIN):
+ def get_room_members(self, room_id):
hs = self.hs
- memberships = yield self.store.get_room_members(
- room_id=room_id, membership=membership
- )
+ users = yield self.store.get_users_in_room(room_id)
- defer.returnValue([hs.parse_userid(m.user_id) for m in memberships])
+ defer.returnValue([hs.parse_userid(u) for u in users])
@defer.inlineCallbacks
def fetch_room_distributions_into(self, room_id, localusers=None,
@@ -531,11 +529,10 @@ class RoomListHandler(BaseHandler):
def get_public_room_list(self):
chunk = yield self.store.get_rooms(is_public=True)
for room in chunk:
- joined_members = yield self.store.get_room_members(
+ joined_users = yield self.store.get_users_in_room(
room_id=room["room_id"],
- membership=Membership.JOIN
)
- room["num_joined_members"] = len(joined_members)
+ room["num_joined_members"] = len(joined_users)
# FIXME (erikj): START is no longer a valid value
defer.returnValue({"start": "START", "end": "END", "chunk": chunk})
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 7d7471d6fe..728d1df8fa 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -434,23 +434,29 @@ class SQLBaseStore(object):
return self.runInteraction("_simple_max_id", func)
- def _get_events(self, event_ids):
+ def _get_events(self, event_ids, check_redacted=True,
+ get_prev_content=False):
return self.runInteraction(
- "_get_events", self._get_events_txn, event_ids
+ "_get_events", self._get_events_txn, event_ids,
+ check_redacted=check_redacted, get_prev_content=get_prev_content,
)
- def _get_events_txn(self, txn, event_ids):
- events = []
- for e_id in event_ids:
- ev = self._get_event_txn(txn, e_id)
+ def _get_events_txn(self, txn, event_ids, check_redacted=True,
+ get_prev_content=False):
+ if not event_ids:
+ return []
- if ev:
- events.append(ev)
-
- return events
+ return [
+ self._get_event_txn(
+ txn, event_id,
+ check_redacted=check_redacted,
+ get_prev_content=get_prev_content
+ )
+ for event_id in event_ids
+ ]
def _get_event_txn(self, txn, event_id, check_redacted=True,
- get_prev_content=True):
+ get_prev_content=False):
sql = (
"SELECT internal_metadata, json, r.event_id FROM event_json as e "
"LEFT JOIN redactions as r ON e.event_id = r.redacts "
@@ -467,6 +473,14 @@ class SQLBaseStore(object):
internal_metadata, js, redacted = res
+ return self._get_event_from_row_txn(
+ txn, internal_metadata, js, redacted,
+ check_redacted=check_redacted,
+ get_prev_content=get_prev_content,
+ )
+
+ def _get_event_from_row_txn(self, txn, internal_metadata, js, redacted,
+ check_redacted=True, get_prev_content=False):
d = json.loads(js)
internal_metadata = json.loads(internal_metadata)
diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py
index 127434d27a..27b7d8eb13 100644
--- a/synapse/storage/roommember.py
+++ b/synapse/storage/roommember.py
@@ -123,6 +123,19 @@ class RoomMemberStore(SQLBaseStore):
else:
return None
+ def get_users_in_room(self, room_id):
+ def f(txn):
+ sql = (
+ "SELECT m.user_id FROM room_memberships as m"
+ " INNER JOIN current_state_events as c"
+ " ON m.event_id = c.event_id"
+ " WHERE m.membership = ? AND m.room_id = ?"
+ )
+
+ txn.execute(sql, (Membership.JOIN, room_id))
+ return [r[0] for r in txn.fetchall()]
+ return self.runInteraction("get_users_in_room", f)
+
def get_room_members(self, room_id, membership=None):
"""Retrieve the current room member list for a room.
@@ -183,20 +196,14 @@ class RoomMemberStore(SQLBaseStore):
)
def _get_members_query_txn(self, txn, where_clause, where_values):
- 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 "
+ "SELECT e.* FROM events as e "
"INNER JOIN room_memberships as m "
"ON e.event_id = m.event_id "
"INNER JOIN current_state_events as c "
"ON m.event_id = c.event_id "
"WHERE %(where)s "
) % {
- "redacted": del_sql,
"where": where_clause,
}
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 6c7fd66933..5327517704 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -15,6 +15,10 @@
from ._base import SQLBaseStore
+import logging
+
+logger = logging.getLogger(__name__)
+
class StateStore(SQLBaseStore):
""" Keeps track of the state at a given event.
@@ -62,14 +66,8 @@ class StateStore(SQLBaseStore):
keyvalues={"state_group": group},
retcol="event_id",
)
- state = []
- for state_id in state_ids:
- s = self._get_events_txn(
- txn,
- [state_id],
- )
- if s:
- state.extend(s)
+
+ state = self._get_events_txn(txn, state_ids)
res[group] = state
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index 8b54aab140..a5e1c38f75 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -137,7 +137,6 @@ class StreamStore(SQLBaseStore):
with_feedback=with_feedback,
)
- @defer.inlineCallbacks
@log_function
def get_room_events_stream(self, user_id, from_key, to_key, room_id,
limit=0, with_feedback=False):
@@ -157,11 +156,6 @@ class StreamStore(SQLBaseStore):
"WHERE m.user_id = ? "
)
- del_sql = (
- "SELECT event_id FROM redactions WHERE redacts = e.event_id "
- "LIMIT 1"
- )
-
if limit:
limit = max(limit, MAX_STREAM_SIZE)
else:
@@ -172,38 +166,42 @@ class StreamStore(SQLBaseStore):
to_id = _parse_stream_token(to_key)
if from_key == to_key:
- defer.returnValue(([], to_key))
- return
+ return defer.succeed(([], to_key))
sql = (
- "SELECT *, (%(redacted)s) AS redacted FROM events AS e WHERE "
+ "SELECT e.event_id, e.stream_ordering FROM events AS e WHERE "
"(e.outlier = 0 AND (room_id IN (%(current)s)) OR "
"(event_id IN (%(invites)s))) "
"AND e.stream_ordering > ? AND e.stream_ordering <= ? "
"ORDER BY stream_ordering ASC LIMIT %(limit)d "
) % {
- "redacted": del_sql,
"current": current_room_membership_sql,
"invites": membership_sql,
"limit": limit
}
- rows = yield self._execute_and_decode(
- sql,
- user_id, user_id, from_id, to_id
- )
+ def f(txn):
+ txn.execute(sql, (user_id, user_id, from_id, to_id,))
- ret = yield self._parse_events(rows)
+ rows = self.cursor_to_dict(txn)
- 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.
- key = to_key
+ ret = self._get_events_txn(
+ txn,
+ [r["event_id"] for r in rows],
+ get_prev_content=True
+ )
+
+ 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.
+ key = to_key
+
+ return ret, key
- defer.returnValue((ret, key))
+ return self.runInteraction("get_room_events_stream", f)
- @defer.inlineCallbacks
@log_function
def paginate_room_events(self, room_id, from_key, to_key=None,
direction='b', limit=-1,
@@ -221,7 +219,9 @@ class StreamStore(SQLBaseStore):
bounds = _get_token_bound(from_key, from_comp)
if to_key:
- bounds = "%s AND %s" % (bounds, _get_token_bound(to_key, to_comp))
+ bounds = "%s AND %s" % (
+ bounds, _get_token_bound(to_key, to_comp)
+ )
if int(limit) > 0:
args.append(int(limit))
@@ -229,87 +229,78 @@ class StreamStore(SQLBaseStore):
else:
limit_str = ""
- del_sql = (
- "SELECT event_id FROM redactions WHERE redacts = events.event_id "
- "LIMIT 1"
- )
-
sql = (
- "SELECT *, (%(redacted)s) AS redacted FROM events"
+ "SELECT * FROM events"
" WHERE outlier = 0 AND room_id = ? AND %(bounds)s"
" ORDER BY topological_ordering %(order)s,"
" stream_ordering %(order)s %(limit)s"
) % {
- "redacted": del_sql,
"bounds": bounds,
"order": order,
"limit": limit_str
}
- rows = yield self._execute_and_decode(
- sql,
- *args
- )
-
- if rows:
- topo = rows[-1]["topological_ordering"]
- toke = rows[-1]["stream_ordering"]
- if direction == 'b':
- topo -= 1
- toke -= 1
- next_token = "t%s-%s" % (topo, toke)
- else:
- # TODO (erikj): We should work out what to do here instead.
- next_token = to_key if to_key else from_key
+ def f(txn):
+ txn.execute(sql, args)
+
+ rows = self.cursor_to_dict(txn)
+
+ if rows:
+ topo = rows[-1]["topological_ordering"]
+ toke = rows[-1]["stream_ordering"]
+ if direction == 'b':
+ topo -= 1
+ toke -= 1
+ next_token = "t%s-%s" % (topo, toke)
+ else:
+ # TODO (erikj): We should work out what to do here instead.
+ next_token = to_key if to_key else from_key
+
+ events = self._get_events_txn(
+ txn,
+ [r["event_id"] for r in rows],
+ get_prev_content=True
+ )
- events = yield self._parse_events(rows)
+ return events, next_token,
- defer.returnValue(
- (
- events,
- next_token
- )
- )
+ return self.runInteraction("paginate_room_events", f)
- @defer.inlineCallbacks
def get_recent_events_for_room(self, room_id, limit, end_token,
with_feedback=False):
# TODO (erikj): Handle compressed feedback
- del_sql = (
- "SELECT event_id FROM redactions WHERE redacts = events.event_id "
- "LIMIT 1"
- )
-
sql = (
- "SELECT *, (%(redacted)s) AS redacted FROM events "
+ "SELECT * FROM events "
"WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0 "
"ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? "
- ) % {
- "redacted": del_sql,
- }
-
- rows = yield self._execute_and_decode(
- sql,
- room_id, end_token, limit
)
- rows.reverse() # As we selected with reverse ordering
+ def f(txn):
+ txn.execute(sql, (room_id, end_token, limit,))
- if rows:
- topo = rows[0]["topological_ordering"]
- toke = rows[0]["stream_ordering"]
- start_token = "t%s-%s" % (topo, toke)
+ rows = self.cursor_to_dict(txn)
- token = (start_token, end_token)
- else:
- token = (end_token, end_token)
+ rows.reverse() # As we selected with reverse ordering
- events = yield self._parse_events(rows)
+ if rows:
+ topo = rows[0]["topological_ordering"]
+ toke = rows[0]["stream_ordering"]
+ start_token = "t%s-%s" % (topo, toke)
+
+ token = (start_token, end_token)
+ else:
+ token = (end_token, end_token)
+
+ events = self._get_events_txn(
+ txn,
+ [r["event_id"] for r in rows],
+ get_prev_content=True
+ )
- ret = (events, token)
+ return events, token
- defer.returnValue(ret)
+ return self.runInteraction("get_recent_events_for_room", f)
def get_room_events_max_id(self):
return self.runInteraction(
|
