diff options
| -rw-r--r-- | synapse/notifier.py | 11 | ||||
| -rw-r--r-- | synapse/replication/resource.py | 2 | ||||
| -rw-r--r-- | synapse/server.py | 13 |
3 files changed, 21 insertions, 5 deletions
diff --git a/synapse/notifier.py b/synapse/notifier.py index d528d1c1e0..054ca59ad2 100644 --- a/synapse/notifier.py +++ b/synapse/notifier.py @@ -143,7 +143,12 @@ class Notifier(object): self.clock = hs.get_clock() self.appservice_handler = hs.get_application_service_handler() - self.federation_sender = hs.get_federation_sender() + + if hs.should_send_federation(): + self.federation_sender = hs.get_federation_sender() + else: + self.federation_sender = None + self.state_handler = hs.get_state_handler() self.clock.looping_call( @@ -220,7 +225,9 @@ class Notifier(object): """Notify any user streams that are interested in this room event""" # poke any interested application service. self.appservice_handler.notify_interested_services(room_stream_id) - self.federation_sender.notify_new_events(room_stream_id) + + if self.federation_sender: + self.federation_sender.notify_new_events(room_stream_id) if event.type == EventTypes.Member and event.membership == Membership.JOIN: self._user_joined_room(event.state_key, event.room_id) diff --git a/synapse/replication/resource.py b/synapse/replication/resource.py index cb9697e378..d79b421cba 100644 --- a/synapse/replication/resource.py +++ b/synapse/replication/resource.py @@ -460,7 +460,7 @@ class ReplicationResource(Resource): ) upto_token = _position_from_rows(to_device_rows, current_position) writer.write_header_and_rows("to_device", to_device_rows, ( - "position", "entity", + "position", "user_id", "device_id", "message_json" ), position=upto_token) @defer.inlineCallbacks diff --git a/synapse/server.py b/synapse/server.py index 6c57ab3e18..ef75ab434c 100644 --- a/synapse/server.py +++ b/synapse/server.py @@ -274,14 +274,23 @@ class HomeServer(object): return TransportLayerClient(self) def build_federation_sender(self): - if self.config.send_federation: + if self.should_send_federation(): return TransactionQueue(self) - else: + elif not self.config.worker_app: return FederationRemoteSendQueue(self) + else: + raise Exception("Workers cannot send federation traffic") def remove_pusher(self, 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 _get(hs): |