diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py
index ee5978da6e..068d4bd087 100644
--- a/synapse/appservice/scheduler.py
+++ b/synapse/appservice/scheduler.py
@@ -132,7 +132,7 @@ class _TransactionController(object):
txn.complete(self.store)
else:
self._start_recoverer(service)
- self.clock.call_later(1000, self.start_polling)
+ self.clock.call_later(1, self.start_polling)
@defer.inlineCallbacks
def on_recovered(self, recoverer):
@@ -202,7 +202,7 @@ class _Recoverer(object):
self.backoff_counter = 1
def recover(self):
- self.clock.call_later(1000 * (2 ** self.backoff_counter), self.retry)
+ self.clock.call_later((2 ** self.backoff_counter), self.retry)
@defer.inlineCallbacks
def retry(self):
diff --git a/synapse/handlers/__init__.py b/synapse/handlers/__init__.py
index 8d345bf936..0c51d615ec 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 (
@@ -54,7 +55,12 @@ 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)
diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index 2c488a46f6..f3cd458e6b 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -26,15 +26,22 @@ import logging
logger = logging.getLogger(__name__)
+def log_failure(failure):
+ logger.error("Application Services Failure: %s", failure.value)
+ logger.error(failure.getTraceback())
+
+
# 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
+ self.scheduler = appservice_scheduler
+ self.started_scheduler = False
@defer.inlineCallbacks
def register(self, app_service):
@@ -90,9 +97,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):
diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index dfce5224a9..6c159b52a0 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -18,7 +18,9 @@ from twisted.internet import defer
from synapse.util.logutils import log_function
from synapse.api.constants import EventTypes
-from .appservice import ApplicationServiceStore
+from .appservice import (
+ ApplicationServiceStore, ApplicationServiceTransactionStore
+)
from .directory import DirectoryStore
from .feedback import FeedbackStore
from .presence import PresenceStore
@@ -79,7 +81,8 @@ class DataStore(RoomMemberStore, RoomStore,
RejectionsStore,
FilteringStore,
PusherStore,
- PushRuleStore
+ PushRuleStore,
+ ApplicationServiceTransactionStore
):
def __init__(self, hs):
|
