End-to-end queued write and wait
@@ -468,7 +468,7 @@ class Node(object):
raise_for_status(r)
return r.json()
- def execute_queued(self, statement, params=None):
+ def execute_queued(self, statement, wait=False, params=None):
body = [statement]
if params is not None:
try:
@@ -476,7 +476,7 @@ class Node(object):
except TypeError:
# Presumably not a list, so append as an object.
body.append(params)
- r = requests.post(self._execute_queued_url(), data=json.dumps([body]))
+ r = requests.post(self._execute_queued_url(wait), data=json.dumps([body]))
raise_for_status(r)
return r.json()
@@ -532,7 +532,10 @@ class Node(object):
if redirect:
rd = "?redirect"
return 'http://' + self.APIAddr() + '/db/execute' + rd
- def _execute_queued_url(self):
+ def _execute_queued_url(self, wait=False):
+ u = '/db/execute?queue'
+ if wait:
+ u = u + '&wait'
return 'http://' + self.APIAddr() + '/db/execute?queue'
def _backup_url(self):
return 'http://' + self.APIAddr() + '/db/backup'
@@ -1207,6 +1210,26 @@ class TestRequestForwarding(unittest.TestCase):
time.sleep(1)
t+=1
+ def test_execute_queued_forward_wait(self):
+ l = self.cluster.wait_for_leader()
+ j = l.execute('CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)')
+ self.assertEqual(j, d_("{'results': [{}]}"))
+
+ f = self.cluster.followers()[0]
+ j = f.execute('INSERT INTO foo(name) VALUES("fiona")')
+ self.assertEqual(j, d_("{'results': [{'last_insert_id': 1, 'rows_affected': 1}]}"))
+ fsmIdx = l.wait_for_all_fsm()
+
+ j = f.execute_queued('INSERT INTO foo(name) VALUES("declan")')
+ self.assertTrue(is_sequence_number(str(j)))
+
+ j = f.execute_queued('INSERT INTO foo(name) VALUES(?)', wait=True, params=["aoife"])
+ self.assertTrue(is_sequence_number(str(j)))
+
+ # Data should be ready immediately, since we waited.
+ j = l.query('SELECT COUNT(*) FROM foo')
+ self.assertEqual(j, d_("{'results': [{'columns': ['COUNT(*)'], 'types': [''], 'values': [[3]]}]}"))
+
class TestEndToEndNonVoter(unittest.TestCase):
def setUp(self):
self.leader = Node(RQLITED_PATH, '0')
