Add stub ApplicationServiceTransactionStore. Bootstrap Recoverers. Fill in stub Transaction functions.

1 files changed, 43 insertions, 7 deletions

diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py
index f54df9c9a5..645d7bf6b2 100644
--- a/synapse/appservice/scheduler.py
+++ b/synapse/appservice/scheduler.py
@@ -71,11 +71,13 @@ class AppServiceScheduler(object):
             clock, store, as_api, self.event_grouper, create_recoverer
         )
 
+    @defer.inlineCallbacks
     def start(self):
         # check for any DOWN ASes and start recoverers for them.
-        _Recoverer.start(
+        recoverers = yield _Recoverer.start(
             self.clock, self.store, self.as_api, self.txn_ctrl.on_recovered
         )
+        self.txn_ctrl.add_recoverers(recoverers)
         self.txn_ctrl.start_polling()
 
     def submit_event_for_as(self, service, event):
@@ -91,12 +93,34 @@ class AppServiceTransaction(object):
         self.events = events
 
     def send(self, as_api):
-        # TODO sends this transaction using this as_api
-        pass
+        """Sends this transaction using the provided AS API interface.
+
+        Args:
+            as_api(ApplicationServiceApi): The API to use to send.
+        Returns:
+            A Deferred which resolves to True if the transaction was sent.
+        """
+        return as_api.push_bulk(
+            service=self.service,
+            events=self.events,
+            txn_id=self.id
+        )
 
     def complete(self, store):
-        # TODO increment txn id on AS and nuke txn contents from db
-        pass
+        """Completes this transaction as successful.
+
+        Marks this transaction ID on the application service and removes the
+        transaction contents from the database.
+
+        Args:
+            store: The database store to operate on.
+        Returns:
+            A Deferred which resolves to True if the transaction was completed.
+        """
+        return store.complete_appservice_txn(
+            service=self.service,
+            txn_id=self.id
+        )
 
 
 class _EventGrouper(object):
@@ -125,6 +149,8 @@ class _TransactionController(object):
         self.as_api = as_api
         self.event_grouper = event_grouper
         self.recoverer_fn = recoverer_fn
+        # keep track of how many recoverers there are
+        self.recoverers = []
 
     def start_polling(self):
         groups = self.event_grouper.drain_groups()
@@ -144,6 +170,10 @@ class _TransactionController(object):
         # TODO mark AS as UP
         pass
 
+    def add_recoverers(self, recoverers):
+        for r in recoverers:
+            self.recoverers.append(r)
+
     def _start_recoverer(self, service):
         recoverer = self.recoverer_fn(service, self.on_recovered)
         recoverer.recover()
@@ -161,9 +191,15 @@ class _TransactionController(object):
 class _Recoverer(object):
 
     @staticmethod
+    @defer.inlineCallbacks
     def start(clock, store, as_api, callback):
-        # TODO check for DOWN ASes and init recoverers
-        pass
+        services = yield store.get_failing_appservices()
+        recoverers = [
+            _Recoverer(clock, store, as_api, s, callback) for s in services
+        ]
+        for r in recoverers:
+            r.recover()
+        defer.returnValue(recoverers)
 
     def __init__(self, clock, store, as_api, service, callback):
         self.clock = clock