Very first cut of calculating actions for events as they come in. Doesn't store them yet. Not very efficient.

2 files changed, 22 insertions, 1 deletions

diff --git a/synapse/handlers/_base.py b/synapse/handlers/_base.py
index 5fd20285d2..a8e8c4f5af 100644
--- a/synapse/handlers/_base.py
+++ b/synapse/handlers/_base.py
@@ -19,9 +19,12 @@ from synapse.api.errors import LimitExceededError, SynapseError, AuthError
 from synapse.crypto.event_signing import add_hashes_and_signatures
 from synapse.api.constants import Membership, EventTypes
 from synapse.types import UserID, RoomAlias
+from synapse.push.action_generator import ActionGenerator
 
 from synapse.util.logcontext import PreserveLoggingContext
 
+from synapse.events.utils import serialize_event
+
 import logging
 
 
@@ -264,6 +267,11 @@ class BaseHandler(object):
             event, context=context
         )
 
+        action_generator = ActionGenerator(self.store)
+        yield action_generator.handle_event(serialize_event(
+            event, self.clock.time_msec()
+        ))
+
         destinations = set(extra_destinations)
         for k, s in context.current_state.items():
             try:
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 2855f2d7c3..18289eb529 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -32,10 +32,12 @@ from synapse.crypto.event_signing import (
 )
 from synapse.types import UserID
 
-from synapse.events.utils import prune_event
+from synapse.events.utils import prune_event, serialize_event
 
 from synapse.util.retryutils import NotRetryingDestination
 
+from synapse.push.action_generator import ActionGenerator
+
 from twisted.internet import defer
 
 import itertools
@@ -1113,6 +1115,11 @@ class FederationHandler(BaseHandler):
             current_state=current_state,
         )
 
+        action_generator = ActionGenerator(self.store)
+        yield action_generator.handle_event(serialize_event(
+            event, self.clock.time_msec())
+        )
+
         defer.returnValue((context, event_stream_id, max_stream_id))
 
     @defer.inlineCallbacks
@@ -1139,6 +1146,12 @@ class FederationHandler(BaseHandler):
             is_new_state=(not outliers and not backfilled),
         )
 
+        for ev_info in event_infos:
+            action_generator = ActionGenerator(self.store)
+            yield action_generator.handle_event(serialize_event(
+                ev_info["event"], self.clock.time_msec())
+            )
+
     @defer.inlineCallbacks
     def _persist_auth_tree(self, auth_events, state, event):
         """Checks the auth chain is valid (and passes auth checks) for the