Ensure only main or federation_sender process can send federation traffic
This commit is contained in:
parent
b69f76c106
commit
26072df6af
|
@ -143,7 +143,12 @@ class Notifier(object):
|
||||||
|
|
||||||
self.clock = hs.get_clock()
|
self.clock = hs.get_clock()
|
||||||
self.appservice_handler = hs.get_application_service_handler()
|
self.appservice_handler = hs.get_application_service_handler()
|
||||||
|
|
||||||
|
if hs.should_send_federation():
|
||||||
self.federation_sender = hs.get_federation_sender()
|
self.federation_sender = hs.get_federation_sender()
|
||||||
|
else:
|
||||||
|
self.federation_sender = None
|
||||||
|
|
||||||
self.state_handler = hs.get_state_handler()
|
self.state_handler = hs.get_state_handler()
|
||||||
|
|
||||||
self.clock.looping_call(
|
self.clock.looping_call(
|
||||||
|
@ -220,6 +225,8 @@ class Notifier(object):
|
||||||
"""Notify any user streams that are interested in this room event"""
|
"""Notify any user streams that are interested in this room event"""
|
||||||
# poke any interested application service.
|
# poke any interested application service.
|
||||||
self.appservice_handler.notify_interested_services(room_stream_id)
|
self.appservice_handler.notify_interested_services(room_stream_id)
|
||||||
|
|
||||||
|
if self.federation_sender:
|
||||||
self.federation_sender.notify_new_events(room_stream_id)
|
self.federation_sender.notify_new_events(room_stream_id)
|
||||||
|
|
||||||
if event.type == EventTypes.Member and event.membership == Membership.JOIN:
|
if event.type == EventTypes.Member and event.membership == Membership.JOIN:
|
||||||
|
|
|
@ -460,7 +460,7 @@ class ReplicationResource(Resource):
|
||||||
)
|
)
|
||||||
upto_token = _position_from_rows(to_device_rows, current_position)
|
upto_token = _position_from_rows(to_device_rows, current_position)
|
||||||
writer.write_header_and_rows("to_device", to_device_rows, (
|
writer.write_header_and_rows("to_device", to_device_rows, (
|
||||||
"position", "entity",
|
"position", "user_id", "device_id", "message_json"
|
||||||
), position=upto_token)
|
), position=upto_token)
|
||||||
|
|
||||||
@defer.inlineCallbacks
|
@defer.inlineCallbacks
|
||||||
|
|
|
@ -274,14 +274,23 @@ class HomeServer(object):
|
||||||
return TransportLayerClient(self)
|
return TransportLayerClient(self)
|
||||||
|
|
||||||
def build_federation_sender(self):
|
def build_federation_sender(self):
|
||||||
if self.config.send_federation:
|
if self.should_send_federation():
|
||||||
return TransactionQueue(self)
|
return TransactionQueue(self)
|
||||||
else:
|
elif not self.config.worker_app:
|
||||||
return FederationRemoteSendQueue(self)
|
return FederationRemoteSendQueue(self)
|
||||||
|
else:
|
||||||
|
raise Exception("Workers cannot send federation traffic")
|
||||||
|
|
||||||
def remove_pusher(self, app_id, push_key, user_id):
|
def remove_pusher(self, app_id, push_key, user_id):
|
||||||
return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
|
return self.get_pusherpool().remove_pusher(app_id, push_key, user_id)
|
||||||
|
|
||||||
|
def should_send_federation(self):
|
||||||
|
"Should this server be sending federation traffic directly?"
|
||||||
|
return self.config.send_federation and (
|
||||||
|
not self.config.worker_app
|
||||||
|
or self.config.worker_app == "synapse.app.federation_sender"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def _make_dependency_method(depname):
|
def _make_dependency_method(depname):
|
||||||
def _get(hs):
|
def _get(hs):
|
||||||
|
|
Loading…
Reference in New Issue