Use seconds; start gluing in the AS scheduler into the AS handler.

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

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)

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

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

tests/appservice/test_scheduler.py

@@ -162,7 +162,7 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase):
         self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count)
         txn.send = Mock(return_value=True)
         # wait for exp backoff
-        self.clock.advance_time(2000)
+        self.clock.advance_time(2)
         self.assertEquals(1, txn.send.call_count)
         self.assertEquals(1, txn.complete.call_count)
         # 2 because it needs to get None to know there are no more txns
@@ -185,21 +185,21 @@ class ApplicationServiceSchedulerRecovererTestCase(unittest.TestCase):
         self.recoverer.recover()
         self.assertEquals(0, self.store.get_oldest_unsent_txn.call_count)
         txn.send = Mock(return_value=False)
-        self.clock.advance_time(2000)
+        self.clock.advance_time(2)
         self.assertEquals(1, txn.send.call_count)
         self.assertEquals(0, txn.complete.call_count)
         self.assertEquals(0, self.callback.call_count)
-        self.clock.advance_time(4000)
+        self.clock.advance_time(4)
         self.assertEquals(2, txn.send.call_count)
         self.assertEquals(0, txn.complete.call_count)
         self.assertEquals(0, self.callback.call_count)
-        self.clock.advance_time(8000)
+        self.clock.advance_time(8)
         self.assertEquals(3, txn.send.call_count)
         self.assertEquals(0, txn.complete.call_count)
         self.assertEquals(0, self.callback.call_count)
         txn.send = Mock(return_value=True)  # successfully send the txn
         pop_txn = True  # returns the txn the first time, then no more.
-        self.clock.advance_time(16000)
+        self.clock.advance_time(16)
         self.assertEquals(1, txn.send.call_count)  # new mock reset call count
         self.assertEquals(1, txn.complete.call_count)
         self.callback.assert_called_once_with(self.recoverer)

tests/handlers/test_appservice.py

@@ -27,10 +27,11 @@ class AppServiceHandlerTestCase(unittest.TestCase):
     def setUp(self):
         self.mock_store = Mock()
         self.mock_as_api = Mock()
+        self.mock_scheduler = Mock()
         hs = Mock()
         hs.get_datastore = Mock(return_value=self.mock_store)
         self.handler = ApplicationServicesHandler(
-            hs, self.mock_as_api
+            hs, self.mock_as_api, self.mock_scheduler
         )
 
     @defer.inlineCallbacks
@@ -52,7 +53,9 @@ class AppServiceHandlerTestCase(unittest.TestCase):
         )
         self.mock_as_api.push = Mock()
         yield self.handler.notify_interested_services(event)
-        self.mock_as_api.push.assert_called_once_with(interested_service, event)
+        self.mock_scheduler.submit_event_for_as.assert_called_once_with(
+            interested_service, event
+        )
 
     @defer.inlineCallbacks
     def test_query_room_alias_exists(self):