Refactoring and fixes

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -246,7 +246,7 @@ def launch_worker(key)
     end
   end
 
-  def clear_workers
+  def clear_finished_workers
     finished_threads = []
     @worker_threads.each do |key, thread|
       next if thread.alive?
@@ -260,32 +260,38 @@ def ensure_workers
     return if free_slots == 0
     keys = @redis.keys(@key)
     keys.shuffle
-    keys = keys[0..free_slots - 1]
     keys.each do |key|
+      next if @worker_threads.has_key?(key)
       @worker_threads[key] = launch_worker(key)
+      free_slots -= 1
+      break if free_slots == 0
     end
   end
 
+  def init_list_pattern_runner
+    @events ||= Queue.new
+    @worker_threads ||= {}
+    @redis ||= connect
+    clear_finished_workers
+    ensure_workers
+  end
+
   # private
   def list_pattern_runner(output_queue)
-    begin
-      @events = Queue.new
-      @worker_threads = {}
-      @redis ||= connect
-      ensure_workers
-    rescue ::Redis::BaseError => e
-      @logger.warn("Redis connection problem", :exception => e)
-      @redis = nil
-    end
+    items = 0
     while !stop?
       begin
-        clear_workers
-        ensure_workers
+        init_list_pattern_runner
         item = @events.pop(true)
         next unless item
         queue_event(item, output_queue)
+        items += 1
       rescue ThreadError => e
+        @logger.warn("Items processed: #{items}")
         sleep(0.1)
+      rescue ::Redis::BaseError => e
+        @logger.warn("Redis connection problem", :exception => e)
+        @redis = nil
       end
     end
   end