From e3e2fc3255665ac888f3e0c0c6e2be39d5fda5f3 Mon Sep 17 00:00:00 2001 From: David Baker Date: Thu, 15 Jan 2015 16:17:21 +0000 Subject: Don't make the pushers' event streams cause people to appear online --- synapse/handlers/events.py | 43 ++++++++++++++++++++++--------------------- 1 file changed, 22 insertions(+), 21 deletions(-) (limited to 'synapse/handlers/events.py') diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index c9ade253dd..54ab27004f 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -47,11 +47,11 @@ class EventStreamHandler(BaseHandler): @defer.inlineCallbacks @log_function def get_stream(self, auth_user_id, pagin_config, timeout=0, - as_client_event=True): + as_client_event=True, affect_presence=True): auth_user = self.hs.parse_userid(auth_user_id) try: - if auth_user not in self._streams_per_user: + if affect_presence and auth_user not in self._streams_per_user: self._streams_per_user[auth_user] = 0 if auth_user in self._stop_timer_per_user: try: @@ -64,7 +64,7 @@ class EventStreamHandler(BaseHandler): yield self.distributor.fire( "started_user_eventstream", auth_user ) - self._streams_per_user[auth_user] += 1 + self._streams_per_user[auth_user] += 1 if pagin_config.from_token is None: pagin_config.from_token = None @@ -92,27 +92,28 @@ class EventStreamHandler(BaseHandler): defer.returnValue(chunk) finally: - self._streams_per_user[auth_user] -= 1 - if not self._streams_per_user[auth_user]: - del self._streams_per_user[auth_user] - - # 10 seconds of grace to allow the client to reconnect again - # before we think they're gone - def _later(): - logger.debug( - "_later stopped_user_eventstream %s", auth_user - ) + if affect_presence: + self._streams_per_user[auth_user] -= 1 + if not self._streams_per_user[auth_user]: + del self._streams_per_user[auth_user] + + # 10 seconds of grace to allow the client to reconnect again + # before we think they're gone + def _later(): + logger.debug( + "_later stopped_user_eventstream %s", auth_user + ) - self._stop_timer_per_user.pop(auth_user, None) + self._stop_timer_per_user.pop(auth_user, None) - yield self.distributor.fire( - "stopped_user_eventstream", auth_user - ) + yield self.distributor.fire( + "stopped_user_eventstream", auth_user + ) - logger.debug("Scheduling _later: for %s", auth_user) - self._stop_timer_per_user[auth_user] = ( - self.clock.call_later(30, _later) - ) + logger.debug("Scheduling _later: for %s", auth_user) + self._stop_timer_per_user[auth_user] = ( + self.clock.call_later(30, _later) + ) class EventHandler(BaseHandler): -- cgit 1.5.1 From 7f6f3f9d6247076493c4e9d48c8282e25892f8b5 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Mon, 26 Jan 2015 16:11:28 +0000 Subject: Pass the current time to serialize event, rather than passing an HS and getting a clock from it and calling time_msec on the clock. Remove the serialize_event method from the HS since it is no longer needed. --- synapse/events/utils.py | 12 ++++++------ synapse/handlers/events.py | 5 ++++- synapse/handlers/message.py | 21 +++++++++++++++------ synapse/handlers/room.py | 7 +++++-- synapse/rest/client/v1/events.py | 8 +++++++- synapse/rest/client/v1/room.py | 9 ++++++++- synapse/server.py | 3 --- 7 files changed, 45 insertions(+), 20 deletions(-) (limited to 'synapse/handlers/events.py') diff --git a/synapse/events/utils.py b/synapse/events/utils.py index bcb5457278..e391aca4cc 100644 --- a/synapse/events/utils.py +++ b/synapse/events/utils.py @@ -89,31 +89,31 @@ def prune_event(event): return type(event)(allowed_fields) -def serialize_event(hs, e, client_event=True): +def serialize_event(e, time_now_ms, client_event=True): # FIXME(erikj): To handle the case of presence events and the like if not isinstance(e, EventBase): return e + time_now_ms = int(time_now_ms) + # Should this strip out None's? d = {k: v for k, v in e.get_dict().items()} if not client_event: # set the age and keep all other keys if "age_ts" in d["unsigned"]: - now = int(hs.get_clock().time_msec()) - d["unsigned"]["age"] = now - d["unsigned"]["age_ts"] + d["unsigned"]["age"] = time_now_ms - d["unsigned"]["age_ts"] return d if "age_ts" in d["unsigned"]: - now = int(hs.get_clock().time_msec()) - d["age"] = now - d["unsigned"]["age_ts"] + d["age"] = time_now_ms - d["unsigned"]["age_ts"] del d["unsigned"]["age_ts"] d["user_id"] = d.pop("sender", None) if "redacted_because" in e.unsigned: d["redacted_because"] = serialize_event( - hs, e.unsigned["redacted_because"] + e.unsigned["redacted_because"], time_now_ms ) del d["unsigned"]["redacted_because"] diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index 01e67b0818..d997917cd6 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -18,6 +18,7 @@ from twisted.internet import defer from synapse.util.logcontext import PreserveLoggingContext from synapse.util.logutils import log_function from synapse.types import UserID +from synapse.events.utils import serialize_event from ._base import BaseHandler @@ -78,8 +79,10 @@ class EventStreamHandler(BaseHandler): auth_user, room_ids, pagin_config, timeout ) + time_now = self.clock.time_msec() + chunks = [ - self.hs.serialize_event(e, as_client_event) for e in events + serialize_event(e, time_now, as_client_event) for e in events ] chunk = { diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py index 6a1104a890..9c3271fe88 100644 --- a/synapse/handlers/message.py +++ b/synapse/handlers/message.py @@ -18,6 +18,7 @@ from twisted.internet import defer from synapse.api.constants import EventTypes, Membership from synapse.api.errors import RoomError from synapse.streams.config import PaginationConfig +from synapse.events.utils import serialize_event from synapse.events.validator import EventValidator from synapse.util.logcontext import PreserveLoggingContext from synapse.types import UserID @@ -100,9 +101,11 @@ class MessageHandler(BaseHandler): "room_key", next_key ) + time_now = self.clock.time_msec() + chunk = { "chunk": [ - self.hs.serialize_event(e, as_client_event) for e in events + serialize_event(e, time_now, as_client_event) for e in events ], "start": pagin_config.from_token.to_string(), "end": next_token.to_string(), @@ -211,7 +214,8 @@ class MessageHandler(BaseHandler): # TODO: This is duplicating logic from snapshot_all_rooms current_state = yield self.state_handler.get_current_state(room_id) - defer.returnValue([self.hs.serialize_event(c) for c in current_state]) + now = self.clock.time_msec() + defer.returnValue([serialize_event(c, now) for c in current_state]) @defer.inlineCallbacks def snapshot_all_rooms(self, user_id=None, pagin_config=None, @@ -283,10 +287,11 @@ class MessageHandler(BaseHandler): start_token = now_token.copy_and_replace("room_key", token[0]) end_token = now_token.copy_and_replace("room_key", token[1]) + time_now = self.clock.time_msec() d["messages"] = { "chunk": [ - self.hs.serialize_event(m, as_client_event) + serialize_event(m, time_now, as_client_event) for m in messages ], "start": start_token.to_string(), @@ -297,7 +302,8 @@ class MessageHandler(BaseHandler): event.room_id ) d["state"] = [ - self.hs.serialize_event(c) for c in current_state + serialize_event(c, time_now, as_client_event) + for c in current_state ] except: logger.exception("Failed to get snapshot") @@ -320,8 +326,9 @@ class MessageHandler(BaseHandler): auth_user = UserID.from_string(user_id) # TODO: These concurrently + time_now = self.clock.time_msec() state_tuples = yield self.state_handler.get_current_state(room_id) - state = [self.hs.serialize_event(x) for x in state_tuples] + state = [serialize_event(x, time_now) for x in state_tuples] member_event = (yield self.store.get_room_member( user_id=user_id, @@ -360,11 +367,13 @@ class MessageHandler(BaseHandler): "Failed to get member presence of %r", m.user_id ) + time_now = self.clock.time_msec() + defer.returnValue({ "membership": member_event.membership, "room_id": room_id, "messages": { - "chunk": [self.hs.serialize_event(m) for m in messages], + "chunk": [serialize_event(m, time_now) for m in messages], "start": start_token.to_string(), "end": end_token.to_string(), }, diff --git a/synapse/handlers/room.py b/synapse/handlers/room.py index edb96cec83..23821d321f 100644 --- a/synapse/handlers/room.py +++ b/synapse/handlers/room.py @@ -16,12 +16,14 @@ """Contains functions for performing events on rooms.""" from twisted.internet import defer +from ._base import BaseHandler + from synapse.types import UserID, RoomAlias, RoomID from synapse.api.constants import EventTypes, Membership, JoinRules from synapse.api.errors import StoreError, SynapseError from synapse.util import stringutils from synapse.util.async import run_on_reactor -from ._base import BaseHandler +from synapse.events.utils import serialize_event import logging @@ -293,8 +295,9 @@ class RoomMemberHandler(BaseHandler): yield self.auth.check_joined_room(room_id, user_id) member_list = yield self.store.get_room_members(room_id=room_id) + time_now = self.clock.time_msec() event_list = [ - self.hs.serialize_event(entry) + serialize_event(entry, time_now) for entry in member_list ] chunk_data = { diff --git a/synapse/rest/client/v1/events.py b/synapse/rest/client/v1/events.py index c69de56863..a0d051227b 100644 --- a/synapse/rest/client/v1/events.py +++ b/synapse/rest/client/v1/events.py @@ -19,6 +19,7 @@ from twisted.internet import defer from synapse.api.errors import SynapseError from synapse.streams.config import PaginationConfig from .base import ClientV1RestServlet, client_path_pattern +from synapse.events.utils import serialize_event import logging @@ -64,14 +65,19 @@ class EventStreamRestServlet(ClientV1RestServlet): class EventRestServlet(ClientV1RestServlet): PATTERN = client_path_pattern("/events/(?P[^/]*)$") + def __init__(self, hs): + super(EventRestServlet, self).__init__(hs) + self.clock = hs.get_clock() + @defer.inlineCallbacks def on_GET(self, request, event_id): auth_user = yield self.auth.get_user_by_req(request) handler = self.handlers.event_handler event = yield handler.get_event(auth_user, event_id) + time_now = self.clock.time_msec() if event: - defer.returnValue((200, self.hs.serialize_event(event))) + defer.returnValue((200, serialize_event(event, time_now))) else: defer.returnValue((404, "Event not found.")) diff --git a/synapse/rest/client/v1/room.py b/synapse/rest/client/v1/room.py index f06e3ddb98..58b09b6fc1 100644 --- a/synapse/rest/client/v1/room.py +++ b/synapse/rest/client/v1/room.py @@ -21,6 +21,7 @@ from synapse.api.errors import SynapseError, Codes from synapse.streams.config import PaginationConfig from synapse.api.constants import EventTypes, Membership from synapse.types import UserID, RoomID, RoomAlias +from synapse.events.utils import serialize_event import json import logging @@ -363,6 +364,10 @@ class RoomInitialSyncRestServlet(ClientV1RestServlet): class RoomTriggerBackfill(ClientV1RestServlet): PATTERN = client_path_pattern("/rooms/(?P[^/]*)/backfill$") + def __init__(self, hs): + super(RoomTriggerBackfill, self).__init__(hs) + self.clock = hs.get_clock() + @defer.inlineCallbacks def on_GET(self, request, room_id): remote_server = urllib.unquote( @@ -374,7 +379,9 @@ class RoomTriggerBackfill(ClientV1RestServlet): handler = self.handlers.federation_handler events = yield handler.backfill(remote_server, room_id, limit) - res = [self.hs.serialize_event(event) for event in events] + time_now = self.clock.time_msec() + + res = [serialize_event(event, time_now) for event in events] defer.returnValue((200, res)) diff --git a/synapse/server.py b/synapse/server.py index 92ed2c5e32..c478f812e6 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -124,9 +124,6 @@ class BaseHomeServer(object): setattr(BaseHomeServer, "get_%s" % (depname), _get) - def serialize_event(self, e, as_client_event=True): - return serialize_event(self, e, as_client_event) - def get_ip_from_request(self, request): # May be an X-Forwarding-For header depending on config ip_addr = request.getClientIP() -- cgit 1.5.1 From 1d779691248be2fcdd43e97352d13c2cef239c10 Mon Sep 17 00:00:00 2001 From: David Baker Date: Tue, 27 Jan 2015 15:58:27 +0000 Subject: Unbreak bad presence merge - don't add these blocks together with an and: they're different things. --- synapse/handlers/events.py | 26 ++++++++++++++------------ 1 file changed, 14 insertions(+), 12 deletions(-) (limited to 'synapse/handlers/events.py') diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index 75fb941008..f1a3e4a4ad 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -51,19 +51,21 @@ class EventStreamHandler(BaseHandler): auth_user = self.hs.parse_userid(auth_user_id) try: - if affect_presence and auth_user not in self._streams_per_user: - self._streams_per_user[auth_user] = 0 - if auth_user in self._stop_timer_per_user: - try: - self.clock.cancel_call_later( - self._stop_timer_per_user.pop(auth_user) + if affect_presence: + if auth_user not in self._streams_per_user: + self._streams_per_user[auth_user] = 0 + if auth_user in self._stop_timer_per_user: + try: + print "cancel",auth_user + self.clock.cancel_call_later( + self._stop_timer_per_user.pop(auth_user) + ) + except: + logger.exception("Failed to cancel event timer") + else: + yield self.distributor.fire( + "started_user_eventstream", auth_user ) - except: - logger.exception("Failed to cancel event timer") - else: - yield self.distributor.fire( - "started_user_eventstream", auth_user - ) self._streams_per_user[auth_user] += 1 if pagin_config.from_token is None: -- cgit 1.5.1 From eba89f093f0376b8499953b53bbd85f9b12b7852 Mon Sep 17 00:00:00 2001 From: David Baker Date: Tue, 27 Jan 2015 16:00:07 +0000 Subject: Need a defer.inlineCallbacks here as we yield in it: otherwise nothing in the cb gets executed. --- synapse/handlers/events.py | 1 + 1 file changed, 1 insertion(+) (limited to 'synapse/handlers/events.py') diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index f1a3e4a4ad..851eebf600 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -99,6 +99,7 @@ class EventStreamHandler(BaseHandler): # 10 seconds of grace to allow the client to reconnect again # before we think they're gone + @defer.inlineCallbacks def _later(): logger.debug( "_later stopped_user_eventstream %s", auth_user -- cgit 1.5.1 From 5eacaeb4a73aa4c4ff450d7b23acabe79124171f Mon Sep 17 00:00:00 2001 From: David Baker Date: Tue, 27 Jan 2015 16:05:23 +0000 Subject: or of course we could just return the deferred --- synapse/handlers/events.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) (limited to 'synapse/handlers/events.py') diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index 851eebf600..48de3630e3 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -99,7 +99,6 @@ class EventStreamHandler(BaseHandler): # 10 seconds of grace to allow the client to reconnect again # before we think they're gone - @defer.inlineCallbacks def _later(): logger.debug( "_later stopped_user_eventstream %s", auth_user @@ -107,7 +106,7 @@ class EventStreamHandler(BaseHandler): self._stop_timer_per_user.pop(auth_user, None) - yield self.distributor.fire( + return self.distributor.fire( "stopped_user_eventstream", auth_user ) -- cgit 1.5.1 From f7c4daa8f94e5968277ba438f6d3da1f0f27ba4c Mon Sep 17 00:00:00 2001 From: David Baker Date: Tue, 27 Jan 2015 16:08:47 +0000 Subject: Oops, remove debugging --- synapse/handlers/events.py | 1 - 1 file changed, 1 deletion(-) (limited to 'synapse/handlers/events.py') diff --git a/synapse/handlers/events.py b/synapse/handlers/events.py index 48de3630e3..52ef07eaae 100644 --- a/synapse/handlers/events.py +++ b/synapse/handlers/events.py @@ -56,7 +56,6 @@ class EventStreamHandler(BaseHandler): self._streams_per_user[auth_user] = 0 if auth_user in self._stop_timer_per_user: try: - print "cancel",auth_user self.clock.cancel_call_later( self._stop_timer_per_user.pop(auth_user) ) -- cgit 1.5.1