summary refs log tree commit diff
path: root/synapse/notifier.py
diff options
context:
space:
mode:
authorErik Johnston <erikj@jki.re>2017-04-04 10:07:57 +0100
committerGitHub <noreply@github.com>2017-04-04 10:07:57 +0100
commit27cc627e425a4a1d6baf3887b435005a571c2271 (patch)
treea5fa797ac7e65f85cae52cccd88451e5672257c6 /synapse/notifier.py
parentMerge pull request #2095 from matrix-org/rav/cull_log_preserves (diff)
parentMerge branch 'develop' of github.com:matrix-org/synapse into erikj/repl_tcp_s... (diff)
downloadsynapse-27cc627e425a4a1d6baf3887b435005a571c2271.tar.xz
Merge pull request #2082 from matrix-org/erikj/repl_tcp_server
Replace HTTP replication with TCP replication (Server side part)
Diffstat (limited to 'synapse/notifier.py')
-rw-r--r--synapse/notifier.py11
1 files changed, 11 insertions, 0 deletions
diff --git a/synapse/notifier.py b/synapse/notifier.py
index 57d6a8cfe3..48566187ab 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -163,6 +163,8 @@ class Notifier(object):
         self.store = hs.get_datastore()
         self.pending_new_room_events = []
 
+        self.replication_callbacks = []
+
         self.clock = hs.get_clock()
         self.appservice_handler = hs.get_application_service_handler()
 
@@ -202,6 +204,12 @@ class Notifier(object):
             lambda: len(self.user_to_user_stream),
         )
 
+    def add_replication_callback(self, cb):
+        """Add a callback that will be called when some new data is available.
+        Callback is not given any arguments.
+        """
+        self.replication_callbacks.append(cb)
+
     def on_new_room_event(self, event, room_stream_id, max_room_stream_id,
                           extra_users=[]):
         """ Used by handlers to inform the notifier something has happened
@@ -507,6 +515,9 @@ class Notifier(object):
             self.replication_deferred = ObservableDeferred(defer.Deferred())
             deferred.callback(None)
 
+        for cb in self.replication_callbacks:
+            preserve_fn(cb)()
+
     @defer.inlineCallbacks
     def wait_for_replication(self, callback, timeout):
         """Wait for an event to happen.