Merge pull request #63 from devdazed/master

Add WhiteListRoundRobinPolicy and attempt to connect to all contact points

cassandra/cluster.py

@@ -1479,7 +1479,8 @@ class ControlConnection(object):
 
         for old_host in self._cluster.metadata.all_hosts():
             if old_host.address != connection.host and \
-                    old_host.address not in found_hosts:
+                    old_host.address not in found_hosts and \
+                    old_host.address not in self._cluster.contact_points:
                 log.debug("[control connection] Found host that has been removed: %r", old_host)
                 self._cluster.remove_host(old_host)
 

cassandra/policies.py

@@ -315,6 +315,53 @@ class TokenAwarePolicy(LoadBalancingPolicy):
         return self._child_policy.on_remove(*args, **kwargs)
 
 
+class WhiteListRoundRobinPolicy(RoundRobinPolicy):
+    """
+    A subclass of :class:`.RoundRobinPolicy` which evenly
+    distributes queries across all nodes in the cluster,
+    regardless of what datacenter the nodes may be in, but
+    only if that node exists in the list of allowed nodes
+
+    This policy is addresses the issue described in
+    https://datastax-oss.atlassian.net/browse/JAVA-145
+    Where connection errors occur when connection
+    attempts are made to private IP addresses remotely
+    """
+    def __init__(self, hosts):
+        """
+        :param hosts: List of hosts
+        """
+        self._allowed_hosts = hosts
+
+    def populate(self, cluster, hosts):
+        self._live_hosts = set()
+        for host in hosts:
+            if host.address in self._allowed_hosts:
+                self._live_hosts.add(host)
+
+        if len(hosts) <= 1:
+            self._position = 0
+        else:
+            self._position = randint(0, len(hosts) - 1)
+
+    def distance(self, host):
+        if host.address in self._allowed_hosts:
+            return HostDistance.LOCAL
+        else:
+            return HostDistance.IGNORED
+
+    def on_up(self, host):
+        if host.address in self._allowed_hosts:
+            self._live_hosts.add(host)
+
+    def on_add(self, host):
+        if host.address in self._allowed_hosts:
+            self._live_hosts.add(host)
+
+    def on_remove(self, host):
+        self._live_hosts.discard(host)
+
+
 class ConvictionPolicy(object):
     """
     A policy which decides when hosts should be considered down