diff --git a/synapse/handlers/__init__.py b/synapse/handlers/__init__.py
index d1b0e032a3..685792dbdc 100644
--- a/synapse/handlers/__init__.py
+++ b/synapse/handlers/__init__.py
@@ -13,6 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from synapse.appservice.scheduler import AppServiceScheduler
from synapse.appservice.api import ApplicationServiceApi
from .register import RegistrationHandler
from .room import (
@@ -56,8 +57,13 @@ class Handlers(object):
self.directory_handler = DirectoryHandler(hs)
self.typing_notification_handler = TypingNotificationHandler(hs)
self.admin_handler = AdminHandler(hs)
+ asapi = ApplicationServiceApi(hs)
self.appservice_handler = ApplicationServicesHandler(
- hs, ApplicationServiceApi(hs)
+ hs, asapi, AppServiceScheduler(
+ clock=hs.get_clock(),
+ store=hs.get_datastore(),
+ as_api=asapi
+ )
)
self.sync_handler = SyncHandler(hs)
self.auth_handler = AuthHandler(hs)
diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index 2c488a46f6..492a630fdc 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -16,57 +16,36 @@
from twisted.internet import defer
from synapse.api.constants import EventTypes, Membership
-from synapse.api.errors import Codes, StoreError, SynapseError
from synapse.appservice import ApplicationService
from synapse.types import UserID
-import synapse.util.stringutils as stringutils
import logging
logger = logging.getLogger(__name__)
+def log_failure(failure):
+ logger.error(
+ "Application Services Failure",
+ exc_info=(
+ failure.type,
+ failure.value,
+ failure.getTracebackObject()
+ )
+ )
+
+
# NB: Purposefully not inheriting BaseHandler since that contains way too much
# setup code which this handler does not need or use. This makes testing a lot
# easier.
class ApplicationServicesHandler(object):
- def __init__(self, hs, appservice_api):
+ def __init__(self, hs, appservice_api, appservice_scheduler):
self.store = hs.get_datastore()
self.hs = hs
self.appservice_api = appservice_api
-
- @defer.inlineCallbacks
- def register(self, app_service):
- logger.info("Register -> %s", app_service)
- # check the token is recognised
- try:
- stored_service = yield self.store.get_app_service_by_token(
- app_service.token
- )
- if not stored_service:
- raise StoreError(404, "Application service not found")
- except StoreError:
- raise SynapseError(
- 403, "Unrecognised application services token. "
- "Consult the home server admin.",
- errcode=Codes.FORBIDDEN
- )
-
- app_service.hs_token = self._generate_hs_token()
-
- # create a sender for this application service which is used when
- # creating rooms, etc..
- account = yield self.hs.get_handlers().registration_handler.register()
- app_service.sender = account[0]
-
- yield self.store.update_app_service(app_service)
- defer.returnValue(app_service)
-
- @defer.inlineCallbacks
- def unregister(self, token):
- logger.info("Unregister as_token=%s", token)
- yield self.store.unregister_app_service(token)
+ self.scheduler = appservice_scheduler
+ self.started_scheduler = False
@defer.inlineCallbacks
def notify_interested_services(self, event):
@@ -90,9 +69,13 @@ class ApplicationServicesHandler(object):
if event.type == EventTypes.Member:
yield self._check_user_exists(event.state_key)
- # Fork off pushes to these services - XXX First cut, best effort
+ if not self.started_scheduler:
+ self.scheduler.start().addErrback(log_failure)
+ self.started_scheduler = True
+
+ # Fork off pushes to these services
for service in services:
- self.appservice_api.push(service, event)
+ self.scheduler.submit_event_for_as(service, event)
@defer.inlineCallbacks
def query_user_exists(self, user_id):
@@ -197,7 +180,14 @@ class ApplicationServicesHandler(object):
return
user_info = yield self.store.get_user_by_id(user_id)
- defer.returnValue(len(user_info) == 0)
+ if len(user_info) > 0:
+ defer.returnValue(False)
+ return
+
+ # user not found; could be the AS though, so check.
+ services = yield self.store.get_app_services()
+ service_list = [s for s in services if s.sender == user_id]
+ defer.returnValue(len(service_list) == 0)
@defer.inlineCallbacks
def _check_user_exists(self, user_id):
@@ -206,6 +196,3 @@ class ApplicationServicesHandler(object):
exists = yield self.query_user_exists(user_id)
defer.returnValue(exists)
defer.returnValue(True)
-
- def _generate_hs_token(self):
- return stringutils.random_string(24)
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 15ba417e06..8aceac28cf 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -201,10 +201,18 @@ class FederationHandler(BaseHandler):
target_user = UserID.from_string(target_user_id)
extra_users.append(target_user)
- yield self.notifier.on_new_room_event(
+ d = self.notifier.on_new_room_event(
event, extra_users=extra_users
)
+ def log_failure(f):
+ logger.warn(
+ "Failed to notify about %s: %s",
+ event.event_id, f.value
+ )
+
+ d.addErrback(log_failure)
+
if event.type == EventTypes.Member:
if event.membership == Membership.JOIN:
user = UserID.from_string(event.state_key)
@@ -427,10 +435,18 @@ class FederationHandler(BaseHandler):
auth_events=auth_events,
)
- yield self.notifier.on_new_room_event(
+ d = self.notifier.on_new_room_event(
new_event, extra_users=[joinee]
)
+ def log_failure(f):
+ logger.warn(
+ "Failed to notify about %s: %s",
+ new_event.event_id, f.value
+ )
+
+ d.addErrback(log_failure)
+
logger.debug("Finished joining %s to %s", joinee, room_id)
finally:
room_queue = self.room_queues[room_id]
@@ -500,10 +516,18 @@ class FederationHandler(BaseHandler):
target_user = UserID.from_string(target_user_id)
extra_users.append(target_user)
- yield self.notifier.on_new_room_event(
+ d = self.notifier.on_new_room_event(
event, extra_users=extra_users
)
+ def log_failure(f):
+ logger.warn(
+ "Failed to notify about %s: %s",
+ event.event_id, f.value
+ )
+
+ d.addErrback(log_failure)
+
if event.type == EventTypes.Member:
if event.content["membership"] == Membership.JOIN:
user = UserID.from_string(event.state_key)
@@ -574,10 +598,18 @@ class FederationHandler(BaseHandler):
)
target_user = UserID.from_string(event.state_key)
- yield self.notifier.on_new_room_event(
+ d = self.notifier.on_new_room_event(
event, extra_users=[target_user],
)
+ def log_failure(f):
+ logger.warn(
+ "Failed to notify about %s: %s",
+ event.event_id, f.value
+ )
+
+ d.addErrback(log_failure)
+
defer.returnValue(event)
@defer.inlineCallbacks
diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py
index 731df00648..bbc7a0f200 100644
--- a/synapse/handlers/presence.py
+++ b/synapse/handlers/presence.py
@@ -33,6 +33,10 @@ logger = logging.getLogger(__name__)
metrics = synapse.metrics.get_metrics_for(__name__)
+# Don't bother bumping "last active" time if it differs by less than 60 seconds
+LAST_ACTIVE_GRANULARITY = 60*1000
+
+
# TODO(paul): Maybe there's one of these I can steal from somewhere
def partition(l, func):
"""Partition the list by the result of func applied to each element."""
@@ -282,6 +286,10 @@ class PresenceHandler(BaseHandler):
if now is None:
now = self.clock.time_msec()
+ prev_state = self._get_or_make_usercache(user)
+ if now - prev_state.state.get("last_active", 0) < LAST_ACTIVE_GRANULARITY:
+ return
+
self.changed_presencelike_data(user, {"last_active": now})
def changed_presencelike_data(self, user, state):
diff --git a/synapse/handlers/typing.py b/synapse/handlers/typing.py
index c2762f92c7..c0b2bd7db0 100644
--- a/synapse/handlers/typing.py
+++ b/synapse/handlers/typing.py
@@ -223,6 +223,7 @@ class TypingNotificationEventSource(object):
def __init__(self, hs):
self.hs = hs
self._handler = None
+ self._room_member_handler = None
def handler(self):
# Avoid cyclic dependency in handler setup
@@ -230,6 +231,11 @@ class TypingNotificationEventSource(object):
self._handler = self.hs.get_handlers().typing_notification_handler
return self._handler
+ def room_member_handler(self):
+ if not self._room_member_handler:
+ self._room_member_handler = self.hs.get_handlers().room_member_handler
+ return self._room_member_handler
+
def _make_event_for(self, room_id):
typing = self.handler()._room_typing[room_id]
return {
@@ -240,19 +246,25 @@ class TypingNotificationEventSource(object):
},
}
+ @defer.inlineCallbacks
def get_new_events_for_user(self, user, from_key, limit):
from_key = int(from_key)
handler = self.handler()
+ joined_room_ids = (
+ yield self.room_member_handler().get_joined_rooms_for_user(user)
+ )
+
events = []
for room_id in handler._room_serials:
+ if room_id not in joined_room_ids:
+ continue
if handler._room_serials[room_id] <= from_key:
continue
- # TODO: check if user is in room
events.append(self._make_event_for(room_id))
- return (events, handler._latest_room_serial)
+ defer.returnValue((events, handler._latest_room_serial))
def get_current_key(self):
return self.handler()._latest_room_serial
|
