summaryrefslogtreecommitdiff
path: root/backends/objectstore.py
diff options
context:
space:
mode:
authordrebs <drebs@leap.se>2013-01-15 18:52:56 -0200
committerdrebs <drebs@leap.se>2013-01-15 18:52:56 -0200
commit984a33de1322437e3979e7324ae042daa52d91e9 (patch)
tree743623729dccfc051905f2415add1db7ab8722f8 /backends/objectstore.py
parentcb22f22b7084b5869fc5c9b5fb8b61f234a07ea8 (diff)
Refactor and fix ObjectStore.
Diffstat (limited to 'backends/objectstore.py')
-rw-r--r--backends/objectstore.py183
1 files changed, 171 insertions, 12 deletions
diff --git a/backends/objectstore.py b/backends/objectstore.py
index 2ab07675..03694532 100644
--- a/backends/objectstore.py
+++ b/backends/objectstore.py
@@ -1,6 +1,5 @@
from u1db.backends import CommonBackend
from u1db import errors, Document, vectorclock
-from leap.soledad import util as soledadutil
class ObjectStore(CommonBackend):
"""
@@ -12,9 +11,9 @@ class ObjectStore(CommonBackend):
# with the database is established in each implementation, so it can
# ensure that u1db data is configured and up-to-date.
self.set_document_factory(Document)
- self._sync_log = soledadutil.SyncLog()
- self._transaction_log = soledadutil.TransactionLog()
- self._conflict_log = soledadutil.ConflictLog(self._factory)
+ self._sync_log = SyncLog()
+ self._transaction_log = TransactionLog()
+ self._conflict_log = ConflictLog(self._factory)
self._replica_uid = replica_uid
self._ensure_u1db_data()
@@ -139,19 +138,18 @@ class ObjectStore(CommonBackend):
def _set_replica_gen_and_trans_id(self, other_replica_uid,
other_generation, other_transaction_id):
- self._get_u1db_data()
+ return self._do_set_replica_gen_and_trans_id(
+ other_replica_uid,
+ other_generation,
+ other_transaction_id)
+
+ def _do_set_replica_gen_and_trans_id(self, other_replica_uid,
+ other_generation, other_transaction_id):
self._sync_log.set_replica_gen_and_trans_id(other_replica_uid,
other_generation,
other_transaction_id)
self._set_u1db_data()
- def _do_set_replica_gen_and_trans_id(self, other_replica_uid,
- other_generation, other_transaction_id):
- return self._set_replica_gen_and_trans_id(
- other_replica_uid,
- other_generation,
- other_transaction_id)
-
def _get_transaction_log(self):
self._get_u1db_data()
return self._transaction_log.get_transaction_log()
@@ -276,3 +274,164 @@ class ObjectStore(CommonBackend):
self._add_conflict(doc.doc_id, my_doc.rev, my_doc.get_json())
doc.has_conflicts = True
self._put_and_update_indexes(my_doc, doc)
+
+
+#----------------------------------------------------------------------------
+# U1DB's Transaction, Sync, and conflict Logs
+#----------------------------------------------------------------------------
+
+class SimpleList(object):
+ def __init__(self):
+ self._log = []
+
+ def _set_log(self, log):
+ self._log = log
+
+ def _get_log(self):
+ return self._log
+
+ log = property(
+ _get_log, _set_log, doc="Log contents.")
+
+ def append(self, msg):
+ self._log.append(msg)
+
+ def reduce(self, func, initializer=None):
+ return reduce(func, self._log, initializer)
+
+ def map(self, func):
+ return map(func, self._get_log())
+
+ def filter(self, func):
+ return filter(func, self._get_log())
+
+
+class TransactionLog(SimpleList):
+ """
+ An ordered list of (generation, doc_id, transaction_id) tuples.
+ """
+
+ def _set_log(self, log):
+ self._log = log
+
+ def _get_log(self):
+ return sorted(self._log, reverse=True)
+
+ log = property(
+ _get_log, _set_log, doc="Log contents.")
+
+ def get_generation(self):
+ """
+ Return the current generation.
+ """
+ gens = self.map(lambda x: x[0])
+ if not gens:
+ return 0
+ return max(gens)
+
+ def get_generation_info(self):
+ """
+ Return the current generation and transaction id.
+ """
+ if not self._log:
+ return(0, '')
+ info = self.map(lambda x: (x[0], x[2]))
+ return reduce(lambda x, y: x if (x[0] > y[0]) else y, info)
+
+ def get_trans_id_for_gen(self, gen):
+ """
+ Get the transaction id corresponding to a particular generation.
+ """
+ log = self.reduce(lambda x, y: y if y[0] == gen else x)
+ if log is None:
+ return None
+ return log[2]
+
+ def whats_changed(self, old_generation):
+ """
+ Return a list of documents that have changed since old_generation.
+ """
+ results = self.filter(lambda x: x[0] > old_generation)
+ seen = set()
+ changes = []
+ newest_trans_id = ''
+ for generation, doc_id, trans_id in results:
+ if doc_id not in seen:
+ changes.append((doc_id, generation, trans_id))
+ seen.add(doc_id)
+ if changes:
+ cur_gen = changes[0][1] # max generation
+ newest_trans_id = changes[0][2]
+ changes.reverse()
+ else:
+ results = self._get_log()
+ if not results:
+ cur_gen = 0
+ newest_trans_id = ''
+ else:
+ cur_gen, _, newest_trans_id = results[0]
+
+ return cur_gen, newest_trans_id, changes
+
+
+ def get_transaction_log(self):
+ """
+ Return only a list of (doc_id, transaction_id)
+ """
+ return map(lambda x: (x[1], x[2]), sorted(self._log))
+
+
+class SyncLog(SimpleList):
+ """
+ A list of (replica_id, generation, transaction_id) tuples.
+ """
+
+ def find_by_replica_uid(self, replica_uid):
+ if not self._get_log():
+ return ()
+ return self.reduce(lambda x, y: y if y[0] == replica_uid else x)
+
+ def get_replica_gen_and_trans_id(self, other_replica_uid):
+ """
+ Return the last known generation and transaction id for the other db
+ replica.
+ """
+ info = self.find_by_replica_uid(other_replica_uid)
+ if not info:
+ return (0, '')
+ return (info[1], info[2])
+
+ def set_replica_gen_and_trans_id(self, other_replica_uid,
+ other_generation, other_transaction_id):
+ """
+ Set the last-known generation and transaction id for the other
+ database replica.
+ """
+ self._log = self.filter(lambda x: x[0] != other_replica_uid)
+ self.append((other_replica_uid, other_generation,
+ other_transaction_id))
+
+class ConflictLog(SimpleList):
+ """
+ A list of (doc_id, my_doc_rev, my_content) tuples.
+ """
+
+ def __init__(self, factory):
+ super(ConflictLog, self).__init__()
+ self._factory = factory
+
+ def delete_conflicts(self, conflicts):
+ for conflict in conflicts:
+ self._log = self.filter(lambda x:
+ x[0] != conflict[0] or x[1] != conflict[1])
+
+ def get_conflicts(self, doc_id):
+ conflicts = self.filter(lambda x: x[0] == doc_id)
+ if not conflicts:
+ return []
+ return reversed(map(lambda x: self._factory(doc_id, x[1], x[2]),
+ conflicts))
+
+ def has_conflicts(self, doc_id):
+ return bool(self.filter(lambda x: x[0] == doc_id))
+