repeated read/write acquisition

Author: technige

neo4j/v1/routing.py

@@ -263,14 +263,32 @@ def refresh_routing_table(self):
     def acquire_for_read(self):
         """ Acquire a connection to a read server.
         """
-        self.refresh_routing_table()
-        return self.acquire(next(self.routing_table.readers))
+        while True:
+            address = None
+            while address is None:
+                self.refresh_routing_table()
+                address = next(self.routing_table.readers)
+            try:
+                connection = self.acquire(address)
+            except ServiceUnavailable:
+                self.remove(address)
+            else:
+                return connection
 
     def acquire_for_write(self):
         """ Acquire a connection to a write server.
         """
-        self.refresh_routing_table()
-        return self.acquire(next(self.routing_table.writers))
+        while True:
+            address = None
+            while address is None:
+                self.refresh_routing_table()
+                address = next(self.routing_table.writers)
+            try:
+                connection = self.acquire(address)
+            except ServiceUnavailable:
+                self.remove(address)
+            else:
+                return connection
 
     def remove(self, address):
         """ Remove an address from the connection pool, if present, closing

test/resources/fail_on_init.script

@@ -0,0 +1,4 @@
+!: AUTO INIT
+!: AUTO RESET
+
+S: <EXIT>

test/resources/router_with_multiple_writers.script

@@ -0,0 +1,8 @@
+!: AUTO INIT
+!: AUTO RESET
+
+C: RUN "CALL dbms.cluster.routing.getServers" {}
+   PULL_ALL
+S: SUCCESS {"fields": ["ttl", "servers"]}
+   RECORD [300, [{"role":"ROUTE","addresses":["127.0.0.1:9001","127.0.0.1:9002","127.0.0.1:9003"]},{"role":"READ","addresses":["127.0.0.1:9004","127.0.0.1:9005"]},{"role":"WRITE","addresses":["127.0.0.1:9006","127.0.0.1:9007"]}]]
+   SUCCESS {}

test/test_routing.py

@@ -577,6 +577,17 @@ def test_connected_to_reader(self):
                 connection = pool.acquire_for_read()
                 assert connection.address in pool.routing_table.readers
 
+    def test_should_retry_if_first_reader_fails(self):
+        with StubCluster({9001: "router.script",
+                          9004: "fail_on_init.script",
+                          9005: "empty.script"}):
+            address = ("127.0.0.1", 9001)
+            with RoutingConnectionPool(connector, address) as pool:
+                assert not pool.routing_table.is_fresh()
+                _ = pool.acquire_for_read()
+                assert ("127.0.0.1", 9004) not in pool.routing_table.readers
+                assert ("127.0.0.1", 9005) in pool.routing_table.readers
+
 
 class RoutingConnectionPoolAcquireForWriteTestCase(ServerTestCase):
 
@@ -596,6 +607,17 @@ def test_connected_to_writer(self):
                 connection = pool.acquire_for_write()
                 assert connection.address in pool.routing_table.writers
 
+    def test_should_retry_if_first_writer_fails(self):
+        with StubCluster({9001: "router_with_multiple_writers.script",
+                          9006: "fail_on_init.script",
+                          9007: "empty.script"}):
+            address = ("127.0.0.1", 9001)
+            with RoutingConnectionPool(connector, address) as pool:
+                assert not pool.routing_table.is_fresh()
+                _ = pool.acquire_for_write()
+                assert ("127.0.0.1", 9006) not in pool.routing_table.writers
+                assert ("127.0.0.1", 9007) in pool.routing_table.writers
+
 
 class RoutingConnectionPoolRemoveTestCase(ServerTestCase):
 

test/util.py

@@ -85,7 +85,6 @@ class ServerTestCase(TestCase):
 
     known_hosts = KNOWN_HOSTS
     known_hosts_backup = known_hosts + ".backup"
-    servers = []
 
     def setUp(self):
         if isfile(self.known_hosts):