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

This commit is contained in:
David Baker 2015-12-10 16:26:08 +00:00
parent 86345a511f
commit 21f135ba76
4 changed files with 81 additions and 1 deletions

View file

@ -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:

View file

@ -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

View file

@ -0,0 +1,47 @@
# -*- coding: utf-8 -*-
# Copyright 2015 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from twisted.internet import defer
import push_rule_evaluator
import logging
logger = logging.getLogger(__name__)
class ActionGenerator:
def __init__(self, store):
self.store = store
# really we want to get all user ids and all profile tags too,
# since we want the actions for each profile tag for every user and
# also actions for a client with no profile tag for each user.
# Currently the event stream doesn't support profile tags on an
# event stream, so we just run the rules for a client with no profile
# tag (ie. we just need all the users).
@defer.inlineCallbacks
def handle_event(self, event):
users = yield self.store.get_users_in_room(event['room_id'])
logger.error("users in room: %r", users)
for uid in users:
evaluator = yield push_rule_evaluator.\
evaluator_for_user_name_and_profile_tag(
uid, None, event['room_id'], self.store
)
actions = yield evaluator.actions_for_event(event)
logger.info("actions for user %s: %s", uid, actions)

View file

@ -291,6 +291,18 @@ class RegistrationStore(SQLBaseStore):
defer.returnValue(ret['user_id'])
defer.returnValue(None)
@defer.inlineCallbacks
def get_all_user_ids(self):
"""Returns all user ids registered on this homeserver"""
return self.runInteraction(
"get_all_user_ids",
self._get_all_user_ids_txn
)
def _get_all_user_ids_txn(self, txn):
txn.execute("SELECT name from users")
return [r[0] for r in txn.fetchall()]
@defer.inlineCallbacks
def count_all_users(self):
"""Counts all users registered on the homeserver."""