Send badge count pushes.

Also fix bugs with retrying.

1 files changed, 19 insertions, 1 deletions

diff --git a/synapse/push/pusherpool.py b/synapse/push/pusherpool.py
index b67ad455ea..7b1ce81e9a 100644
--- a/synapse/push/pusherpool.py
+++ b/synapse/push/pusherpool.py
@@ -126,11 +126,29 @@ class PusherPool:
             for u in users_affected:
                 if u in self.pushers:
                     for p in self.pushers[u].values():
-                        p.on_new_notifications(min_stream_id, max_stream_id)
+                        yield p.on_new_notifications(min_stream_id, max_stream_id)
         except:
             logger.exception("Exception in pusher on_new_notifications")
 
     @defer.inlineCallbacks
+    def on_new_receipts(self, min_stream_id, max_stream_id, affected_room_ids):
+        yield run_on_reactor()
+        try:
+            # Need to subtract 1 from the minimum because the lower bound here
+            # is not inclusive
+            updated_receipts = yield self.store.get_all_updated_receipts(
+                min_stream_id - 1, max_stream_id
+            )
+            # This returns a tuple, user_id is at index 3
+            users_affected = set([r[3] for r in updated_receipts])
+            for u in users_affected:
+                if u in self.pushers:
+                    for p in self.pushers[u].values():
+                        yield p.on_new_receipts(min_stream_id, max_stream_id)
+        except:
+            logger.exception("Exception in pusher on_new_receipts")
+
+    @defer.inlineCallbacks
     def _refresh_pusher(self, app_id, pushkey, user_id):
         resultlist = yield self.store.get_pushers_by_app_id_and_pushkey(
             app_id, pushkey