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.
|
