[CLIENT] Synchronize access to the connections collection and mutation of @current instance variable

Author: estolfo

elasticsearch-transport/lib/elasticsearch/transport/transport/connections/selector.rb

@@ -61,18 +61,30 @@ def select(options={})
           class RoundRobin
             include Base
 
+            # @option arguments [Connections::Collection] :connections Collection with connections.
+            #
+            def initialize(arguments = {})
+              super
+              @mutex = Mutex.new
+              @current = nil
+            end
+
             # Returns the next connection from the collection, rotating them in round-robin fashion.
             #
             # @return [Connections::Connection]
             #
             def select(options={})
-              # On Ruby 1.9, Array#rotate could be used instead
-              @current = !defined?(@current) || @current.nil? ? 0 : @current+1
-              @current = 0 if @current >= connections.size
-              connections[@current]
+              @mutex.synchronize do
+                conns = connections
+                if @current && (@current < conns.size-1)
+                  @current += 1
+                else
+                  @current = 0
+                end
+                conns[@current]
+              end
             end
           end
-
         end
       end
     end

elasticsearch-transport/spec/elasticsearch/connections/collection_spec.rb

@@ -228,7 +228,7 @@
     context 'when multiple threads are used' do
 
       let(:connections) do
-        100.times.collect do |i|
+        20.times.collect do |i|
           Elasticsearch::Transport::Transport::Connections::Connection.new(host: { host: i })
         end
       end
@@ -238,13 +238,16 @@
       end
 
       it 'allows threads to select connections in parallel' do
-        threads = []
-        threads << Thread.new do
-          2000.times do |i|
-            collection.get_connection
+        expect(10.times.collect do
+          threads = []
+          20.times do
+            threads << Thread.new do
+              collection.get_connection
+            end
           end
-        end
-        threads.collect {|t| t.join}
+          threads.collect { |t| t.join }
+          collection.get_connection.host[:host]
+        end).to all(eq(0))
       end
     end
   end