Connection reload threadsafety + more efficient manticore pooling

elasticsearch-transport/lib/elasticsearch/transport/transport/base.rb

@@ -28,6 +28,8 @@ module Base
         # @see Client#initialize
         #
         def initialize(arguments={}, &block)
+          @state_mutex = Mutex.new
+
           @hosts       = arguments[:hosts]   || []
           @options     = arguments[:options] || {}
           @block       = block
@@ -94,10 +96,12 @@ def resurrect_dead_connections!
         # @api private
         #
         def __rebuild_connections(arguments={})
-          @hosts       = arguments[:hosts]    || []
-          @options     = arguments[:options]  || {}
-          __close_connections
-          @connections = __build_connections
+          @state_mutex.synchronize do
+            @hosts       = arguments[:hosts]    || []
+            @options     = arguments[:options]  || {}
+            __close_connections
+            @connections = __build_connections
+          end
         end
 
         # Closes the connections collection.

elasticsearch-transport/lib/elasticsearch/transport/transport/http/manticore.rb

@@ -45,6 +45,19 @@ module HTTP
         class Manticore
           include Base
 
+          def initialize(arguments={}, &block)
+            @manticore = build_client(arguments[:options] || {})
+            super()
+          end
+
+          # Should just be run once at startup
+          def build_client(options={})
+            client_options = options[:transport_options] || {}
+            client_options[:ssl] = options[:ssl] || {}
+
+            @manticore = ::Manticore::Client.new(client_options)
+          end
+
           # Performs the request by invoking {Transport::Base#perform_request} with a block.
           #
           # @return [Response]
@@ -84,9 +97,6 @@ def __build_connections
               @request_options[:headers] = options[:headers]
             end
 
-            client_options = options[:transport_options] || {}
-            client_options[:ssl] = options[:ssl] || {}
-
             Connections::Collection.new \
               :connections => hosts.map { |host|
                 host[:protocol]   = host[:scheme] || DEFAULT_PROTOCOL
@@ -95,11 +105,9 @@ def __build_connections
                 host.delete(:user)     # auth is not supported here.
                 host.delete(:password) # use the headers
 
-                url               = __full_url(host)
-
                 Connections::Connection.new \
                   :host => host,
-                  :connection => ::Manticore::Client.new(client_options)
+                  :connection => @manticore
               },
               :selector_class => options[:selector_class],
               :selector => options[:selector]