Introducing DummyConnectionPool for when only 1 connection defined

Changelog.rst

@@ -12,6 +12,8 @@ Changelog
  * added support for proper SSL certificate handling
  * Required parameters are now checked for non-empty values
  * ConnectionPool now checks if any connections were defined
+ * DummyConnectionPool introduced when no load balancing is needed (only one
+   connection defined)
  * fixed a race condition in ConnectionPool
 
 1.2.0 (2014-08-03)

elasticsearch/connection_pool.py

@@ -227,6 +227,22 @@ class ConnectionPool(object):
         # only one connection, no need for a selector
         return connections[0]
 
-        return connection
+
+class DummyConnectionPool(ConnectionPool):
+    def __init__(self, connections, **kwargs):
+        if len(connections) != 1:
+            raise ImproperlyConfigured("DummyConnectionPool needs exactly one "
+                    "connection defined.")
+        # we need connection opts for sniffing logic
+        self.connection_opts = connections
+        self.connection = connections[0][0]
+        self.connections = (self.connection, )
+
+    def get_connection(self):
+        return self.connection
+
+    def _noop(self, *args, **kwargs):
+        pass
+    mark_dead = mark_live = resurrect = _noop
 
 

elasticsearch/transport.py

@@ -2,7 +2,7 @@ import re
 import time
 
 from .connection import Urllib3HttpConnection
-from .connection_pool import ConnectionPool
+from .connection_pool import ConnectionPool, DummyConnectionPool
 from .serializer import JSONSerializer, Deserializer, DEFAULT_SERIALIZERS
 from .exceptions import ConnectionError, TransportError, SerializationError, \
                         ConnectionTimeout, ImproperlyConfigured
@@ -165,8 +165,12 @@ class Transport(object):
             return self.connection_class(**kwargs)
         connections = map(_create_connection, hosts)
 
-        # pass the hosts dicts to the connection pool to optionally extract parameters from
-        self.connection_pool = self.connection_pool_class(list(zip(connections, hosts)), **self.kwargs)
+        connections = list(zip(connections, hosts))
+        if len(connections) == 1:
+            self.connection_pool = DummyConnectionPool(connections)
+        else:
+            # pass the hosts dicts to the connection pool to optionally extract parameters from
+            self.connection_pool = self.connection_pool_class(connections, **self.kwargs)
 
     def get_connection(self):
         """

test_elasticsearch/run_tests.py

@@ -8,9 +8,6 @@ import subprocess
 
 import nose
 
-from test_elasticsearch.test_server import get_client
-from test_elasticsearch.test_cases import SkipTest
-
 def fetch_es_repo():
     # user is manually setting YAML dir, don't tamper with it
     if 'TEST_ES_YAML_DIR' in environ:
@@ -35,6 +32,9 @@ def fetch_es_repo():
     if environ.get('TEST_ES_NOFETCH', False):
         return
 
+    from test_elasticsearch.test_server import get_client
+    from test_elasticsearch.test_cases import SkipTest
+
     # find out the sha of the running es
     try:
         es = get_client()

test_elasticsearch/test_connection_pool.py

@@ -1,11 +1,15 @@
 import time
 
-from elasticsearch.connection_pool import ConnectionPool, RoundRobinSelector
+from elasticsearch.connection_pool import ConnectionPool, RoundRobinSelector, DummyConnectionPool
 from elasticsearch.exceptions import ImproperlyConfigured
 
 from .test_cases import TestCase
 
 class TestConnectionPool(TestCase):
+    def test_dummy_cp_raises_exception_on_more_connections(self):
+        self.assertRaises(ImproperlyConfigured, DummyConnectionPool, [])
+        self.assertRaises(ImproperlyConfigured, DummyConnectionPool, [object(), object()])
+
     def test_raises_exception_when_no_connections_defined(self):
         self.assertRaises(ImproperlyConfigured, ConnectionPool, [])
 

test_elasticsearch/test_transport.py

@@ -4,6 +4,7 @@ import time
 
 from elasticsearch.transport import Transport, get_host_info
 from elasticsearch.connection import Connection, ThriftConnection
+from elasticsearch.connection_pool import DummyConnectionPool
 from elasticsearch.exceptions import ConnectionError, ImproperlyConfigured
 
 from .test_cases import TestCase
@@ -52,6 +53,12 @@ class TestHostsInfoCallback(TestCase):
 
 
 class TestTransport(TestCase):
+    def test_single_connection_uses_dummy_connection_pool(self):
+        t = Transport([{}])
+        self.assertIsInstance(t.connection_pool, DummyConnectionPool)
+        t = Transport([{'host': 'localhost'}])
+        self.assertIsInstance(t.connection_pool, DummyConnectionPool)
+
     def test_host_with_scheme_different_from_connection_fails(self):
         self.assertRaises(ImproperlyConfigured, Transport, [{'host': 'localhost', 'scheme': 'thrift'}])
         self.assertRaises(ImproperlyConfigured, Transport, [{'host': 'localhost', 'scheme': 'http'}], connection_class=ThriftConnection)
@@ -100,7 +107,7 @@ class TestTransport(TestCase):
 
     def test_kwargs_passed_on_to_connection_pool(self):
         dt = object()
-        t = Transport([{}], dead_timeout=dt)
+        t = Transport([{}, {}], dead_timeout=dt)
         self.assertIs(dt, t.connection_pool.dead_timeout)
 
     def test_custom_connection_class(self):
@@ -125,19 +132,21 @@ class TestTransport(TestCase):
         self.assertEquals(4, len(t.get_connection().calls))
 
     def test_failed_connection_will_be_marked_as_dead(self):
-        t = Transport([{'exception': ConnectionError('abandon ship')}], connection_class=DummyConnection)
+        t = Transport([{'exception': ConnectionError('abandon ship')}] * 2, connection_class=DummyConnection)
 
         self.assertRaises(ConnectionError, t.perform_request, 'GET', '/')
         self.assertEquals(0, len(t.connection_pool.connections))
 
     def test_resurrected_connection_will_be_marked_as_live_on_success(self):
-        t = Transport([{}], connection_class=DummyConnection)
-        con = t.connection_pool.get_connection()
-        t.connection_pool.mark_dead(con)
+        t = Transport([{}, {}], connection_class=DummyConnection)
+        con1 = t.connection_pool.get_connection()
+        con2 = t.connection_pool.get_connection()
+        t.connection_pool.mark_dead(con1)
+        t.connection_pool.mark_dead(con2)
 
         t.perform_request('GET', '/')
         self.assertEquals(1, len(t.connection_pool.connections))
-        self.assertEquals(0, len(t.connection_pool.dead_count))
+        self.assertEquals(1, len(t.connection_pool.dead_count))
 
     def test_sniff_will_use_seed_connections(self):
         t = Transport([{'data': CLUSTER_NODES}], connection_class=DummyConnection)