Refactor a bit and fix off-by-one

Author: Juuso Mäyränen

lib/logstash/inputs/redis.rb

@@ -260,11 +260,11 @@ def list_pattern_stop
   end
 
   # private
-  def launch_worker(output_queue, key)
+  def worker_consume(output_queue, key)
     @logger.debug("Launched worker for #{key}")
     redis = new_redis_instance
     begin
-      (0..@max_items_per_worker).each do
+      (0..@max_items_per_worker-1).each do
         if stop?
           @logger.debug("Breaking from thread #{key} as it was requested to stop")
           break
@@ -285,21 +285,26 @@ def threadpool_capacity?
     @threadpool.remaining_capacity > 0
   end
 
+  # private
+  def launch_worker(output_queue, key)
+    @current_workers.add(key)
+    @threadpool.post do
+      begin
+        worker_consume(output_queue, key)
+      ensure
+        @current_workers.delete(key)
+      end
+    end
+  end
+
   # private
   def ensure_workers(output_queue)
     return unless threadpool_capacity?
     keys = @redis.keys(@key)
     keys.shuffle
     keys.each do |key|
       next if @current_workers.include?(key)
-      @current_workers.add(key)
-      @threadpool.post do
-        begin
-          launch_worker(output_queue, key)
-        ensure
-          @current_workers.delete(key)
-        end
-      end
+      launch_worker(output_queue, key)
       break unless threadpool_capacity?
     end
   end