* global pop timeout

* ThreadedPool
* add connection factory

pairing with @tenderlove and @therealadam

Author: hone

lib/resque.rb

@@ -17,11 +17,17 @@
 require 'resque/coder'
 require 'resque/multi_json_coder'
 require 'resque/consumer'
+require 'resque/threaded_pool'
 
 module Resque
   include Helpers
   extend self
 
+  @consumer_timeout = 5
+  class << self
+    attr_accessor :consumer_timeout
+  end
+
   # Accepts:
   #   1. A 'hostname:port' String
   #   2. A 'hostname:port:db' String (to select the Redis db)
@@ -30,6 +36,14 @@ module Resque
   #   5. An instance of `Redis`, `Redis::Client`, `Redis::DistRedis`,
   #      or `Redis::Namespace`.
   def redis=(server)
+    @server = server
+    @redis  = create_connection(server)
+    @queues = Hash.new { |h,name|
+      h[name] = Resque::Queue.new(name, @redis, coder)
+    }
+  end
+
+  def create_connection(server)
     case server
     when String
       if server =~ /redis\:\/\//
@@ -42,15 +56,16 @@ def redis=(server)
       end
       namespace ||= :resque
 
-      @redis = Redis::Namespace.new(namespace, :redis => redis)
+      Redis::Namespace.new(namespace, :redis => redis)
     when Redis::Namespace
-      @redis = server
+      server
     else
-      @redis = Redis::Namespace.new(:resque, :redis => server)
+      Redis::Namespace.new(:resque, :redis => server)
     end
-    @queues = Hash.new { |h,name|
-      h[name] = Resque::Queue.new(name, @redis, coder)
-    }
+  end
+
+  def server
+    @server
   end
 
   # Encapsulation of encode/decode. Overwrite this to use it across Resque.

lib/resque/consumer.rb

@@ -15,7 +15,7 @@ def await
       end
     end
 
-    def initialize(queue, timeout = 5)
+    def initialize(queue, timeout = Resque.consumer_timeout)
       @queue           = queue
       @should_pause    = false
       @should_shutdown = false

lib/resque/threaded_pool.rb

@@ -0,0 +1,26 @@
+module Resque
+  class ThreadedPool
+
+    def initialize(queue, pool)
+      @queue = queue
+      @pool  = pool
+      @threads = []
+      @consumers = []
+    end
+
+
+    def start
+      @consumers.clear
+      @threads = @pool.times.map {
+        c = Consumer.new(@queue)
+        @consumers << c
+        Thread.new { c.consume }
+      }
+    end
+
+    def stop
+      @consumers.each { |c| c.shutdown }
+      @threads.each { |t| t.join }
+    end
+  end
+end

test/threaded_pool_test.rb

@@ -0,0 +1,42 @@
+require "test_helper"
+
+Thread.abort_on_exception = true
+
+module Resque
+
+  describe "ThreadedPool" do
+
+    class Actionable
+      @@ran = []
+
+      def self.ran
+        @@ran
+      end
+
+      def run
+        self.class.ran << self
+      end
+    end
+
+    before do
+      @write  = Queue.new(:foo)
+      conn = Resque.create_connection(Resque.server)
+      @read  = Queue.new(:foo, conn)
+      @tp = ThreadedPool.new(@read, 5)
+    end
+
+    it "processes work" do
+      Resque.consumer_timeout = 1
+      5.times { @write << Actionable.new }
+      @tp.start
+      sleep 1
+      @tp.stop
+      assert @write.empty?
+    end
+
+    it "recovers from blowed-up jobs" do
+      skip
+    end
+
+  end
+end