diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index a83c03da9a..84941eef45 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -15,7 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import logging
-from typing import Optional
+from typing import Optional, Tuple
from six import iteritems, itervalues, string_types
@@ -42,6 +42,7 @@ from synapse.api.errors import (
)
from synapse.api.room_versions import KNOWN_ROOM_VERSIONS, RoomVersions
from synapse.api.urls import ConsentURIBuilder
+from synapse.events import EventBase
from synapse.events.validator import EventValidator
from synapse.logging.context import run_in_background
from synapse.metrics.background_process_metrics import run_as_background_process
@@ -365,10 +366,13 @@ class EventCreationHandler(object):
self.notifier = hs.get_notifier()
self.config = hs.config
self.require_membership_for_aliases = hs.config.require_membership_for_aliases
+ self._is_event_writer = (
+ self.config.worker.writers.events == hs.get_instance_name()
+ )
self.room_invite_state_types = self.hs.config.room_invite_state_types
- self.send_event_to_master = ReplicationSendEventRestServlet.make_client(hs)
+ self.send_event = ReplicationSendEventRestServlet.make_client(hs)
# This is only used to get at ratelimit function, and maybe_kick_guest_users
self.base_handler = BaseHandler(hs)
@@ -632,7 +636,9 @@ class EventCreationHandler(object):
msg = self._block_events_without_consent_error % {"consent_uri": consent_uri}
raise ConsentNotGivenError(msg=msg, consent_uri=consent_uri)
- async def send_nonmember_event(self, requester, event, context, ratelimit=True):
+ async def send_nonmember_event(
+ self, requester, event, context, ratelimit=True
+ ) -> int:
"""
Persists and notifies local clients and federation of an event.
@@ -641,6 +647,9 @@ class EventCreationHandler(object):
context (Context) the context of the event.
ratelimit (bool): Whether to rate limit this send.
is_guest (bool): Whether the sender is a guest.
+
+ Return:
+ The stream_id of the persisted event.
"""
if event.type == EventTypes.Member:
raise SynapseError(
@@ -661,7 +670,7 @@ class EventCreationHandler(object):
)
return prev_state
- await self.handle_new_client_event(
+ return await self.handle_new_client_event(
requester=requester, event=event, context=context, ratelimit=ratelimit
)
@@ -690,7 +699,7 @@ class EventCreationHandler(object):
async def create_and_send_nonmember_event(
self, requester, event_dict, ratelimit=True, txn_id=None
- ):
+ ) -> Tuple[EventBase, int]:
"""
Creates an event, then sends it.
@@ -713,10 +722,10 @@ class EventCreationHandler(object):
spam_error = "Spam is not permitted here"
raise SynapseError(403, spam_error, Codes.FORBIDDEN)
- await self.send_nonmember_event(
+ stream_id = await self.send_nonmember_event(
requester, event, context, ratelimit=ratelimit
)
- return event
+ return event, stream_id
@measure_func("create_new_client_event")
@defer.inlineCallbacks
@@ -776,7 +785,7 @@ class EventCreationHandler(object):
@measure_func("handle_new_client_event")
async def handle_new_client_event(
self, requester, event, context, ratelimit=True, extra_users=[]
- ):
+ ) -> int:
"""Processes a new event. This includes checking auth, persisting it,
notifying users, sending to remote servers, etc.
@@ -789,6 +798,9 @@ class EventCreationHandler(object):
context (EventContext)
ratelimit (bool)
extra_users (list(UserID)): Any extra users to notify about event
+
+ Return:
+ The stream_id of the persisted event.
"""
if event.is_state() and (event.type, event.state_key) == (
@@ -828,8 +840,9 @@ class EventCreationHandler(object):
success = False
try:
# If we're a worker we need to hit out to the master.
- if self._is_worker_app:
- await self.send_event_to_master(
+ if not self._is_event_writer:
+ result = await self.send_event(
+ instance_name=self.config.worker.writers.events,
event_id=event.event_id,
store=self.store,
requester=requester,
@@ -838,14 +851,17 @@ class EventCreationHandler(object):
ratelimit=ratelimit,
extra_users=extra_users,
)
+ stream_id = result["stream_id"]
+ event.internal_metadata.stream_ordering = stream_id
success = True
- return
+ return stream_id
- await self.persist_and_notify_client_event(
+ stream_id = await self.persist_and_notify_client_event(
requester, event, context, ratelimit=ratelimit, extra_users=extra_users
)
success = True
+ return stream_id
finally:
if not success:
# Ensure that we actually remove the entries in the push actions
@@ -888,13 +904,13 @@ class EventCreationHandler(object):
async def persist_and_notify_client_event(
self, requester, event, context, ratelimit=True, extra_users=[]
- ):
+ ) -> int:
"""Called when we have fully built the event, have already
calculated the push actions for the event, and checked auth.
- This should only be run on master.
+ This should only be run on the instance in charge of persisting events.
"""
- assert not self._is_worker_app
+ assert self._is_event_writer
if ratelimit:
# We check if this is a room admin redacting an event so that we
@@ -1078,6 +1094,8 @@ class EventCreationHandler(object):
# matters as sometimes presence code can take a while.
run_in_background(self._bump_active_time, requester.user)
+ return event_stream_id
+
async def _bump_active_time(self, user):
try:
presence = self.hs.get_presence_handler()
|
