From 836c6d19339d7ace076cc106044f0abc4426ecce Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Kr=C3=A1l?= Date: Sat, 20 Dec 2014 00:35:24 +0100 Subject: [PATCH] Introducing DummyConnectionPool for when only 1 connection defined --- Changelog.rst | 2 ++ elasticsearch/connection_pool.py | 18 +++++++++++++++++- elasticsearch/transport.py | 10 +++++++--- test_elasticsearch/run_tests.py | 6 +++--- test_elasticsearch/test_connection_pool.py | 6 +++++- test_elasticsearch/test_transport.py | 21 +++++++++++++++------ 6 files changed, 49 insertions(+), 14 deletions(-) diff --git a/Changelog.rst b/Changelog.rst index 876ad0fe..87b10397 100644 --- a/Changelog.rst +++ b/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) diff --git a/elasticsearch/connection_pool.py b/elasticsearch/connection_pool.py index 169f21da..2991a61b 100644 --- a/elasticsearch/connection_pool.py +++ b/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 diff --git a/elasticsearch/transport.py b/elasticsearch/transport.py index 27e9b9ba..6f005bdb 100644 --- a/elasticsearch/transport.py +++ b/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): """ diff --git a/test_elasticsearch/run_tests.py b/test_elasticsearch/run_tests.py index bb86b18d..43178902 100755 --- a/test_elasticsearch/run_tests.py +++ b/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() diff --git a/test_elasticsearch/test_connection_pool.py b/test_elasticsearch/test_connection_pool.py index 5489acc9..fdbe16fd 100644 --- a/test_elasticsearch/test_connection_pool.py +++ b/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, []) diff --git a/test_elasticsearch/test_transport.py b/test_elasticsearch/test_transport.py index 0556f4d3..9e4c2ddb 100644 --- a/test_elasticsearch/test_transport.py +++ b/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)