Finish synapse.appservice.scheduler implementation.

With tests to assert behaviour. Not hooked up yet. Stub datastore methods
not implemented yet.

3 files changed, 73 insertions, 51 deletions

diff --git a/synapse/appservice/__init__.py b/synapse/appservice/__init__.py
index cc6c381566..743a8278ad 100644
--- a/synapse/appservice/__init__.py
+++ b/synapse/appservice/__init__.py
@@ -25,6 +25,45 @@ class ApplicationServiceState(object):
     UP = "up"
 
 
+class AppServiceTransaction(object):
+    """Represents an application service transaction."""
+
+    def __init__(self, service, id, events):
+        self.service = service
+        self.id = id
+        self.events = events
+
+    def send(self, as_api):
+        """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):
+        """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 ApplicationService(object):
     """Defines an application service. This definition is mostly what is
     provided to the /register AS API.
diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py
index 2b3aa3b0ea..50ad3b8e83 100644
--- a/synapse/appservice/scheduler.py
+++ b/synapse/appservice/scheduler.py
@@ -88,45 +88,6 @@ class AppServiceScheduler(object):
         self.event_grouper.on_receive(service, event)
 
 
-class AppServiceTransaction(object):
-    """Represents an application service transaction."""
-
-    def __init__(self, service, id, events):
-        self.service = service
-        self.id = id
-        self.events = events
-
-    def send(self, as_api):
-        """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):
-        """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):
     """Groups events for the same application service together.
     """
@@ -156,14 +117,18 @@ class _TransactionController(object):
         # keep track of how many recoverers there are
         self.recoverers = []
 
+    @defer.inlineCallbacks
     def start_polling(self):
         groups = self.event_grouper.drain_groups()
         for service in groups:
-            txn_id = self._get_next_txn_id(service)
-            txn = AppServiceTransaction(service, txn_id, groups[service])
-            self._store_txn(txn)
-            if self._is_service_up(service):
-                if txn.send(self.as_api):
+            txn = yield self.store.create_appservice_txn(
+                service=service,
+                events=groups[service]
+            )
+            service_is_up = yield self._is_service_up(service)
+            if service_is_up:
+                sent = yield txn.send(self.as_api)
+                if sent:
                     txn.complete(self.store)
                 else:
                     self._start_recoverer(service)
@@ -207,14 +172,10 @@ class _TransactionController(object):
             logger.error("Failed to apply appservice state DOWN to service %s",
                          service)
 
+    @defer.inlineCallbacks
     def _is_service_up(self, service):
-        pass
-
-    def _get_next_txn_id(self, service):
-        pass  # TODO work out the next txn_id for this service
-
-    def _store_txn(self, txn):
-        pass
+        state = yield self.store.get_appservice_state(service)
+        defer.returnValue(state == ApplicationServiceState.UP)
 
 
 class _Recoverer(object):
diff --git a/synapse/storage/appservice.py b/synapse/storage/appservice.py
index 214f6d99c5..6fde7dcc66 100644
--- a/synapse/storage/appservice.py
+++ b/synapse/storage/appservice.py
@@ -354,6 +354,16 @@ class ApplicationServiceTransactionStore(SQLBaseStore):
         """
         pass
 
+    def get_appservice_state(self, service):
+        """Get the application service state.
+
+        Args:
+            service(ApplicationService): The service whose state to set.
+        Returns:
+            A Deferred which resolves to ApplicationServiceState.
+        """
+        pass
+
     def set_appservice_state(self, service, state):
         """Set the application service state.
 
@@ -365,6 +375,18 @@ class ApplicationServiceTransactionStore(SQLBaseStore):
         """
         pass
 
+    def create_appservice_txn(self, service, events):
+        """Atomically creates a new transaction for this application service
+        with the given list of events.
+
+        Args:
+            service(ApplicationService): The service who the transaction is for.
+            events(list<Event>): A list of events to put in the transaction.
+        Returns:
+            ApplicationServiceTransaction: A new transaction.
+        """
+        pass
+
     def complete_appservice_txn(self, txn_id, service):
         """Completes an application service transaction.