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

Merge branch 'develop' into markjh/edu_frequency

Browse source
This commit is contained in:
Mark Haines 2015-12-01 19:15:27 +00:00
commit f73ea0bda2
45 changed files with 854 additions and 252 deletions
Download patch file Download diff file Expand all files Collapse all files

21
synapse/handlers/account_data.py
View file

@ -29,9 +29,10 @@ class AccountDataEventSource(object):
last_stream_id = from_key
current_stream_id = yield self.store.get_max_account_data_stream_id()
tags = yield self.store.get_updated_tags(user_id, last_stream_id)
results = []
tags = yield self.store.get_updated_tags(user_id, last_stream_id)
for room_id, room_tags in tags.items():
results.append({
"type": "m.tag",
@ -39,6 +40,24 @@ class AccountDataEventSource(object):
"room_id": room_id,
})
account_data, room_account_data = (
yield self.store.get_updated_account_data_for_user(user_id, last_stream_id)
)
for account_data_type, content in account_data.items():
results.append({
"type": account_data_type,
"content": content,
})
for room_id, account_data in room_account_data.items():
for account_data_type, content in account_data.items():
results.append({
"type": account_data_type,
"content": content,
"room_id": room_id,
})
defer.returnValue((results, current_stream_id))
@defer.inlineCallbacks

40
synapse/handlers/message.py
View file

@ -359,6 +359,10 @@ class MessageHandler(BaseHandler):
tags_by_room = yield self.store.get_tags_for_user(user_id)
account_data, account_data_by_room = (
yield self.store.get_account_data_for_user(user_id)
)
public_room_ids = yield self.store.get_public_room_ids()
limit = pagin_config.limit
@ -436,14 +440,22 @@ class MessageHandler(BaseHandler):
for c in current_state.values()
]
account_data = []
account_data_events = []
tags = tags_by_room.get(event.room_id)
if tags:
account_data.append({
account_data_events.append({
"type": "m.tag",
"content": {"tags": tags},
})
d["account_data"] = account_data
account_data = account_data_by_room.get(event.room_id, {})
for account_data_type, content in account_data.items():
account_data_events.append({
"type": account_data_type,
"content": content,
})
d["account_data"] = account_data_events
except:
logger.exception("Failed to get snapshot")
@ -456,9 +468,17 @@ class MessageHandler(BaseHandler):
consumeErrors=True
).addErrback(unwrapFirstError)
account_data_events = []
for account_data_type, content in account_data.items():
account_data_events.append({
"type": account_data_type,
"content": content,
})
ret = {
"rooms": rooms_ret,
"presence": presence,
"account_data": account_data_events,
"receipts": receipt,
"end": now_token.to_string(),
}
@ -498,14 +518,22 @@ class MessageHandler(BaseHandler):
user_id, room_id, pagin_config, membership, member_event_id, is_guest
)
account_data = []
account_data_events = []
tags = yield self.store.get_tags_for_room(user_id, room_id)
if tags:
account_data.append({
account_data_events.append({
"type": "m.tag",
"content": {"tags": tags},
})
result["account_data"] = account_data
account_data = yield self.store.get_account_data_for_room(user_id, room_id)
for account_data_type, content in account_data.items():
account_data_events.append({
"type": account_data_type,
"content": content,
})
result["account_data"] = account_data_events
defer.returnValue(result)

2
synapse/handlers/presence.py
View file

@ -467,7 +467,7 @@ class PresenceHandler(BaseHandler):
)
@defer.inlineCallbacks
def send_invite(self, observer_user, observed_user):
def send_presence_invite(self, observer_user, observed_user):
"""Request the presence of a local or remote user for a local user"""
if not self.hs.is_mine(observer_user):
raise SynapseError(400, "User is not hosted on this Home Server")

147
synapse/handlers/search.py
View file

@ -131,6 +131,17 @@ class SearchHandler(BaseHandler):
if batch_group == "room_id":
room_ids.intersection_update({batch_group_key})
if not room_ids:
defer.returnValue({
"search_categories": {
"room_events": {
"results": {},
"count": 0,
"highlights": [],
}
}
})
rank_map = {} # event_id -> rank of event
allowed_events = []
room_groups = {} # Holds result of grouping by room, if applicable
@ -178,85 +189,67 @@ class SearchHandler(BaseHandler):
s["results"].append(e.event_id)
elif order_by == "recent":
# In this case we specifically loop through each room as the given
# limit applies to each room, rather than a global list.
# This is not necessarilly a good idea.
for room_id in room_ids:
room_events = []
if batch_group == "room_id" and batch_group_key == room_id:
pagination_token = batch_token
else:
room_events = []
i = 0
pagination_token = batch_token
# We keep looping and we keep filtering until we reach the limit
# or we run out of things.
# But only go around 5 times since otherwise synapse will be sad.
while len(room_events) < search_filter.limit() and i < 5:
i += 1
search_result = yield self.store.search_rooms(
room_ids, search_term, keys, search_filter.limit() * 2,
pagination_token=pagination_token,
)
if search_result["highlights"]:
highlights.update(search_result["highlights"])
results = search_result["results"]
results_map = {r["event"].event_id: r for r in results}
rank_map.update({r["event"].event_id: r["rank"] for r in results})
filtered_events = search_filter.filter([
r["event"] for r in results
])
events = yield self._filter_events_for_client(
user.to_string(), filtered_events
)
room_events.extend(events)
room_events = room_events[:search_filter.limit()]
if len(results) < search_filter.limit() * 2:
pagination_token = None
i = 0
# We keep looping and we keep filtering until we reach the limit
# or we run out of things.
# But only go around 5 times since otherwise synapse will be sad.
while len(room_events) < search_filter.limit() and i < 5:
i += 1
search_result = yield self.store.search_room(
room_id, search_term, keys, search_filter.limit() * 2,
pagination_token=pagination_token,
)
if search_result["highlights"]:
highlights.update(search_result["highlights"])
results = search_result["results"]
results_map = {r["event"].event_id: r for r in results}
rank_map.update({r["event"].event_id: r["rank"] for r in results})
filtered_events = search_filter.filter([
r["event"] for r in results
])
events = yield self._filter_events_for_client(
user.to_string(), filtered_events
)
room_events.extend(events)
room_events = room_events[:search_filter.limit()]
if len(results) < search_filter.limit() * 2:
pagination_token = None
break
else:
pagination_token = results[-1]["pagination_token"]
if room_events:
res = results_map[room_events[-1].event_id]
pagination_token = res["pagination_token"]
group = room_groups.setdefault(room_id, {})
if pagination_token:
next_batch = encode_base64("%s\n%s\n%s" % (
"room_id", room_id, pagination_token
))
group["next_batch"] = next_batch
if batch_token:
global_next_batch = next_batch
group["results"] = [e.event_id for e in room_events]
group["order"] = max(
e.origin_server_ts/1000 for e in room_events
if hasattr(e, "origin_server_ts")
)
allowed_events.extend(room_events)
# Normalize the group orders
if room_groups:
if len(room_groups) > 1:
mx = max(g["order"] for g in room_groups.values())
mn = min(g["order"] for g in room_groups.values())
for g in room_groups.values():
g["order"] = (g["order"] - mn) * 1.0 / (mx - mn)
break
else:
room_groups.values()[0]["order"] = 1
pagination_token = results[-1]["pagination_token"]
for event in room_events:
group = room_groups.setdefault(event.room_id, {
"results": [],
})
group["results"].append(event.event_id)
if room_events and len(room_events) >= search_filter.limit():
last_event_id = room_events[-1].event_id
pagination_token = results_map[last_event_id]["pagination_token"]
global_next_batch = encode_base64("%s\n%s\n%s" % (
"all", "", pagination_token
))
for room_id, group in room_groups.items():
group["next_batch"] = encode_base64("%s\n%s\n%s" % (
"room_id", room_id, pagination_token
))
allowed_events.extend(room_events)
else:
# We should never get here due to the guard earlier.

72
synapse/handlers/sync.py
View file

@ -100,6 +100,7 @@ class InvitedSyncResult(collections.namedtuple("InvitedSyncResult", [
class SyncResult(collections.namedtuple("SyncResult", [
"next_batch", # Token for the next sync
"presence", # List of presence events for the user.
"account_data", # List of account_data events for the user.
"joined", # JoinedSyncResult for each joined room.
"invited", # InvitedSyncResult for each invited room.
"archived", # ArchivedSyncResult for each archived room.
@ -195,6 +196,12 @@ class SyncHandler(BaseHandler):
)
)
account_data, account_data_by_room = (
yield self.store.get_account_data_for_user(
sync_config.user.to_string()
)
)
tags_by_room = yield self.store.get_tags_for_user(
sync_config.user.to_string()
)
@ -211,6 +218,7 @@ class SyncHandler(BaseHandler):
timeline_since_token=timeline_since_token,
ephemeral_by_room=ephemeral_by_room,
tags_by_room=tags_by_room,
account_data_by_room=account_data_by_room,
)
joined.append(room_sync)
elif event.membership == Membership.INVITE:
@ -230,11 +238,13 @@ class SyncHandler(BaseHandler):
leave_token=leave_token,
timeline_since_token=timeline_since_token,
tags_by_room=tags_by_room,
account_data_by_room=account_data_by_room,
)
archived.append(room_sync)
defer.returnValue(SyncResult(
presence=presence,
account_data=self.account_data_for_user(account_data),
joined=joined,
invited=invited,
archived=archived,
@ -244,7 +254,8 @@ class SyncHandler(BaseHandler):
@defer.inlineCallbacks
def full_state_sync_for_joined_room(self, room_id, sync_config,
now_token, timeline_since_token,
ephemeral_by_room, tags_by_room):
ephemeral_by_room, tags_by_room,
account_data_by_room):
"""Sync a room for a client which is starting without any state
Returns:
A Deferred JoinedSyncResult.
@ -262,19 +273,38 @@ class SyncHandler(BaseHandler):
state=current_state,
ephemeral=ephemeral_by_room.get(room_id, []),
account_data=self.account_data_for_room(
room_id, tags_by_room
room_id, tags_by_room, account_data_by_room
),
))
def account_data_for_room(self, room_id, tags_by_room):
account_data = []
def account_data_for_user(self, account_data):
account_data_events = []
for account_data_type, content in account_data.items():
account_data_events.append({
"type": account_data_type,
"content": content,
})
return account_data_events
def account_data_for_room(self, room_id, tags_by_room, account_data_by_room):
account_data_events = []
tags = tags_by_room.get(room_id)
if tags is not None:
account_data.append({
account_data_events.append({
"type": "m.tag",
"content": {"tags": tags},
})
return account_data
account_data = account_data_by_room.get(room_id, {})
for account_data_type, content in account_data.items():
account_data_events.append({
"type": account_data_type,
"content": content,
})
return account_data_events
@defer.inlineCallbacks
def ephemeral_by_room(self, sync_config, now_token, since_token=None):
@ -341,7 +371,8 @@ class SyncHandler(BaseHandler):
@defer.inlineCallbacks
def full_state_sync_for_archived_room(self, room_id, sync_config,
leave_event_id, leave_token,
timeline_since_token, tags_by_room):
timeline_since_token, tags_by_room,
account_data_by_room):
"""Sync a room for a client which is starting without any state
Returns:
A Deferred JoinedSyncResult.
@ -358,7 +389,7 @@ class SyncHandler(BaseHandler):
timeline=batch,
state=leave_state,
account_data=self.account_data_for_room(
room_id, tags_by_room
room_id, tags_by_room, account_data_by_room
),
))
@ -415,6 +446,13 @@ class SyncHandler(BaseHandler):
since_token.account_data_key,
)
account_data, account_data_by_room = (
yield self.store.get_updated_account_data_for_user(
sync_config.user.to_string(),
since_token.account_data_key,
)
)
joined = []
archived = []
if len(room_events) <= timeline_limit:
@ -469,7 +507,7 @@ class SyncHandler(BaseHandler):
state=state,
ephemeral=ephemeral_by_room.get(room_id, []),
account_data=self.account_data_for_room(
room_id, tags_by_room
room_id, tags_by_room, account_data_by_room
),
)
logger.debug("Result for room %s: %r", room_id, room_sync)
@ -492,14 +530,15 @@ class SyncHandler(BaseHandler):
for room_id in joined_room_ids:
room_sync = yield self.incremental_sync_with_gap_for_room(
room_id, sync_config, since_token, now_token,
ephemeral_by_room, tags_by_room
ephemeral_by_room, tags_by_room, account_data_by_room
)
if room_sync:
joined.append(room_sync)
for leave_event in leave_events:
room_sync = yield self.incremental_sync_for_archived_room(
sync_config, leave_event, since_token, tags_by_room
sync_config, leave_event, since_token, tags_by_room,
account_data_by_room
)
archived.append(room_sync)
@ -510,6 +549,7 @@ class SyncHandler(BaseHandler):
defer.returnValue(SyncResult(
presence=presence,
account_data=self.account_data_for_user(account_data),
joined=joined,
invited=invited,
archived=archived,
@ -566,7 +606,8 @@ class SyncHandler(BaseHandler):
@defer.inlineCallbacks
def incremental_sync_with_gap_for_room(self, room_id, sync_config,
since_token, now_token,
ephemeral_by_room, tags_by_room):
ephemeral_by_room, tags_by_room,
account_data_by_room):
""" Get the incremental delta needed to bring the client up to date for
the room. Gives the client the most recent events and the changes to
state.
@ -606,7 +647,7 @@ class SyncHandler(BaseHandler):
state=state,
ephemeral=ephemeral_by_room.get(room_id, []),
account_data=self.account_data_for_room(
room_id, tags_by_room
room_id, tags_by_room, account_data_by_room
),
)
@ -616,7 +657,8 @@ class SyncHandler(BaseHandler):
@defer.inlineCallbacks
def incremental_sync_for_archived_room(self, sync_config, leave_event,
since_token, tags_by_room):
since_token, tags_by_room,
account_data_by_room):
""" Get the incremental delta needed to bring the client up to date for
the archived room.
Returns:
@ -654,7 +696,7 @@ class SyncHandler(BaseHandler):
timeline=batch,
state=state_events_delta,
account_data=self.account_data_for_room(
leave_event.room_id, tags_by_room
leave_event.room_id, tags_by_room, account_data_by_room
),
)