Fixed variables names.

Reworked strategy class.
Reorked and added tests.
Some small cleanup.

Author: denis-ignatenko

README.rst

@@ -88,11 +88,10 @@ How to use the connector
     con = tarantool.Connection(host='localhost',
                                port=3301,
                                user='test',
-                               password='test',
-                               connect_now=True)
+                               password='test')
 
 
-    resp = con.eval('return "Hello Wrold!"')
+    resp = con.eval('return "Hello World!"')
     if resp.data[0] == "Hello World!":
         print('Ok')
 

tarantool/connection.py

@@ -378,7 +378,8 @@ def call(self, func_name, *args):
             args = args[0]
 
         request = RequestCall(self, func_name, args, self.call_16)
-        return self._send_request(request)
+        response = self._send_request(request)
+        return response
 
     def eval(self, expr, *args):
         '''
@@ -396,7 +397,7 @@ def eval(self, expr, *args):
         # This allows to use a tuple or list as an argument
         if len(args) == 1 and isinstance(args[0], (list, tuple)):
             args = args[0]
-        
+
         request = RequestEval(self, expr, args)
         return self._send_request(request)
 

tarantool/const.py

@@ -87,4 +87,4 @@
 # Default delay between attempts to reconnect (seconds)
 RECONNECT_DELAY = 0.1
 # Default cluster nodes list refresh interval (seconds)
-DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS = 60000
+CLUSTER_DISCOVERY_DELAY = 60

tarantool/mesh_connection.py

@@ -5,6 +5,8 @@
 '''
 
 import time
+from itertools import cycle
+
 from tarantool.connection import Connection
 from tarantool.error import NetworkError
 from tarantool.utils import ENCODING_DEFAULT
@@ -13,7 +15,7 @@
     SOCKET_TIMEOUT,
     RECONNECT_MAX_ATTEMPTS,
     RECONNECT_DELAY,
-    DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS,
+    CLUSTER_DISCOVERY_DELAY,
 )
 
 from tarantool.request import (
@@ -23,47 +25,52 @@
 
 class RoundRobinStrategy(object):
     """
-    Simple roundrobin address rotation
+    Simple round-robin address rotation
     """
     def __init__(self, addrs):
-        self.addrs = addrs
-        self.pos = 0
+        self.reload(addrs)
+
+    def reload(self, addrs):
+        self.addrs = []
+        for i in addrs:
+            if i.get("host", None) and i.get("port", None) \
+                        and isinstance(i["port"], int):
+                if i in self.addrs:
+                    continue
+                self.addrs.append(i)
+        self.__gen = cycle(self.addrs)
 
     def getnext(self):
-        tmp = self.pos
-        self.pos = (self.pos + 1) % len(self.addrs)
-        return self.addrs[tmp]
+        return next(self.__gen)
 
 
 def parse_uri(uri_str):
-    if not uri_str or ':' not in uri_str:
-        return None
-    uri = uri_str.split(':')
+    if not uri_str or uri_str.count(":") != 1:
+        return
+    uri = uri_str.split(':', 1)
     host = uri[0]
     try:
         port = int(uri[1])
     except ValueError:
-        return None
-    
+        return
+
     if host and port:
         return {'host': host, 'port': port}
-    else:
-        return None
 
 
 class MeshConnection(Connection):
     '''
-    Represents a connection to a cluster of Tarantool servers.
+   Represents a connection to a cluster of Tarantool servers.
 
-    This class uses Connection to connect to one of the nodes of the cluster.
-    The initial list of nodes is passed to the constructor in 'addrs' parameter.
-    The class set in 'strategy_class' parameter is used to select a node from
-    the list and switch nodes in case of unavailability of the current node.
+   This class uses Connection to connect to one of the nodes of the cluster.
+   The initial list of nodes is passed to the constructor in 'addrs' parameter.
+   The class set in 'strategy_class' parameter is used to select a node from
+   the list and switch nodes in case of unavailability of the current node.
 
-    'get_nodes_function_name' param of the constructor sets the name of a stored
-    Lua function used to refresh the list of available nodes. The function takes
-    no parameters and returns a list of strings in format 'host:port'. A generic
-    function for getting the list of nodes looks like this:
+   'cluster_discovery_function' param of the constructor sets the name of a stored
+   Lua function used to refresh the list of available nodes. The function takes
+   no parameters and returns a list of strings in format 'host:port'. A generic
+   function for getting the list of nodes looks like this:
 
     .. code-block:: lua
 
@@ -116,29 +123,27 @@ def __init__(self, host=None, port=None,
                  connection_timeout=CONNECTION_TIMEOUT,
                  addrs=None,
                  strategy_class=RoundRobinStrategy,
-                 get_nodes_function_name=None,
-                 nodes_refresh_interval=DEFAULT_CLUSTER_DISCOVERY_DELAY_MILLIS):
+                 cluster_discovery_function=None,
+                 cluster_discovery_delay=CLUSTER_DISCOVERY_DELAY):
 
-        addrs_list = []
+        if addrs is None:
+            addrs = []
 
         if host and port:
-            addrs_list.append({'host':host, 'port':port})
-
-        if addrs:
-            for addr in addrs:
-                if 'host' in addr and 'port' in addr:
-                    addrs_list.append({'host': addr['host'], 'port': addr['port']})
+            addrs.insert(0, {'host': host, 'port': port})
 
         self.strategy_class = strategy_class
-        self.strategy = strategy_class(addrs_list)
-
-        if not host and not port:
-            addr = self.strategy.getnext()
-            host = addr['host']
-            port = addr['port']
-        
-        self.get_nodes_function_name = get_nodes_function_name
-        self.nodes_refresh_interval = nodes_refresh_interval
+        self.strategy = strategy_class(addrs)
+
+        if not self.strategy.addrs and connect_now:
+            raise ValueError("Host/port or addrs list must be set")
+
+        addr = self.strategy.getnext()
+        host = addr['host']
+        port = addr['port']
+
+        self.cluster_discovery_function = cluster_discovery_function
+        self.cluster_discovery_delay = cluster_discovery_delay
         self.last_nodes_refresh = time.time()
         super(MeshConnection, self).__init__(host=host,
                                              port=port,
@@ -152,67 +157,75 @@ def __init__(self, host=None, port=None,
                                              call_16=call_16,
                                              connection_timeout=connection_timeout)
 
+    def _opt_reconnect(self):
+        '''
+        Use original opt_reconnect with address rotation
+        '''
+
+        for i in range(len(self.strategy.addrs)+1):
+            try:
+                super(MeshConnection, self)._opt_reconnect()
+            except NetworkError:
+                addr = self.strategy.getnext()
+                self.host = addr["host"]
+                self.port = addr["port"]
+                continue
+
+        if not self._socket:
+            raise NetworkError
+
     def _opt_refresh_instances(self):
-        """
+        '''
         Refresh list of cluster instances.
         If current connection not in server list will change connection.
-        """
+        '''
         now = time.time()
 
-        if self.connected and now - self.last_nodes_refresh > self.nodes_refresh_interval/1000:
-            request = RequestCall(self, self.get_nodes_function_name, (), self.call_16)
+        if not self.connected:
+            return
+
+        if now - self.last_nodes_refresh > self.cluster_discovery_delay:
+            request = RequestCall(self,
+                                  self.cluster_discovery_function,
+                                  (),
+                                  self.call_16)
+
             resp = self._send_request_wo_reconnect(request)
 
             # got data to refresh
             if resp.data and resp.data[0]:
-                addrs = list(parse_uri(i) for i in resp.data[0])
-                self.strategy = self.strategy_class(addrs)
-                self.last_nodes_refresh = now
-
-            if {'host': self.host, 'port': self.port} not in addrs:
-                addr = self.strategy.getnext()
-                self.host = addr['host']
-                self.port = addr['port']
-                self.close()
-
-        if not self.connected:
-
-            nattempts = (len(self.strategy.addrs) * 2) + 1
-
-            while nattempts >= 0:
-                try:
+                addrs = []
+                for i in resp.data[0]:
+                    addr = parse_uri(i)
+                    if addr:
+                        addrs.append(addr)
+                if addrs:
+                    self.strategy.reload(addrs)
+                    self.last_nodes_refresh = now
+                else:
+                    raise NetworkError('No addresses to connect')
+
+                # Check current addr in list
+                if {'host': self.host, 'port': self.port} not in addrs:
+                    self.close()
                     addr = self.strategy.getnext()
-                    if addr['host'] != self.host or addr['port'] != self.port:
-                        self.host = addr['host']
-                        self.port = addr['port']
-                        self._opt_reconnect()
-                        break
-                    else:
-                        nattempts -= 1
-                except NetworkError:
-                    continue
-            else:
-                raise NetworkError
+                    self.host = addr['host']
+                    self.port = addr['port']
+                    self._opt_reconnect()
 
     def _send_request(self, request):
         '''
         Send the request to the server through the socket.
         Return an instance of `Response` class.
 
-        Update instances list from server `get_nodes_function_name` function.
+        Update instances list from server `cluster_discovery_function` function.
 
         :param request: object representing a request
         :type request: `Request` instance
 
         :rtype: `Response` instance
         '''
-        if self.get_nodes_function_name:
+        if self.cluster_discovery_function:
             self._opt_refresh_instances()
 
-        try:
-            return super(MeshConnection, self)._send_request(request)
-        except NetworkError:
-            self.connected = False
-            self._opt_refresh_instances()
-        finally:
-            return super(MeshConnection, self)._send_request(request)
+        return super(MeshConnection, self)._send_request(request)