From 2cc998fed879357376edb35d5088d88a078dd576 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Thu, 20 Jul 2017 17:13:18 +0100
Subject: Fix replication. And notify

---
 synapse/replication/tcp/streams.py | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

(limited to 'synapse/replication/tcp')

diff --git a/synapse/replication/tcp/streams.py b/synapse/replication/tcp/streams.py
index fbafe12cc2..4c60bf79f9 100644
--- a/synapse/replication/tcp/streams.py
+++ b/synapse/replication/tcp/streams.py
@@ -118,6 +118,12 @@ CurrentStateDeltaStreamRow = namedtuple("CurrentStateDeltaStream", (
     "state_key",  # str
     "event_id",  # str, optional
 ))
+GroupsStreamRow = namedtuple("GroupsStreamRow", (
+    "group_id",  # str
+    "user_id",  # str
+    "type",  # str
+    "content",  # dict
+))
 
 
 class Stream(object):
@@ -464,6 +470,19 @@ class CurrentStateDeltaStream(Stream):
         super(CurrentStateDeltaStream, self).__init__(hs)
 
 
+class GroupServerStream(Stream):
+    NAME = "groups"
+    ROW_TYPE = GroupsStreamRow
+
+    def __init__(self, hs):
+        store = hs.get_datastore()
+
+        self.current_token = store.get_group_stream_token
+        self.update_function = store.get_all_groups_changes
+
+        super(GroupServerStream, self).__init__(hs)
+
+
 STREAMS_MAP = {
     stream.NAME: stream
     for stream in (
@@ -482,5 +501,6 @@ STREAMS_MAP = {
         TagAccountDataStream,
         AccountDataStream,
         CurrentStateDeltaStream,
+        GroupServerStream,
     )
 }
-- 
cgit 1.4.1


From f807f7f80442aec48a5e1f6b6b6f1a88e707b1e2 Mon Sep 17 00:00:00 2001
From: hera <matrix@template.upcloud.com>
Date: Thu, 12 Oct 2017 10:50:44 +0000
Subject: log when we get an exception handling replication updates

---
 synapse/replication/tcp/resource.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

(limited to 'synapse/replication/tcp')

diff --git a/synapse/replication/tcp/resource.py b/synapse/replication/tcp/resource.py
index 3ea3ca5a6f..6c1beca4e3 100644
--- a/synapse/replication/tcp/resource.py
+++ b/synapse/replication/tcp/resource.py
@@ -160,7 +160,11 @@ class ReplicationStreamer(object):
                             "Getting stream: %s: %s -> %s",
                             stream.NAME, stream.last_token, stream.upto_token
                         )
-                        updates, current_token = yield stream.get_updates()
+                        try:
+                            updates, current_token = yield stream.get_updates()
+                        except:
+                            logger.info("Failed to handle stream %s", stream.NAME)
+                            raise
 
                         logger.debug(
                             "Sending %d updates to %d connections",
-- 
cgit 1.4.1