Allow connection option to be callable

This change allows :connection option to be callable and avoid calling
start on bunny sessions.

lib/sneakers/publisher.rb

@@ -6,9 +6,6 @@ class Publisher
     def initialize(opts = {})
       @mutex = Mutex.new
       @opts = Sneakers::CONFIG.merge(opts)
-      # If we've already got a bunny object, use it.  This allows people to
-      # specify all kinds of options we don't need to know about (e.g. for ssl).
-      @bunny = @opts[:connection]
     end
 
     def publish(msg, options = {})
@@ -29,8 +26,15 @@ def ensure_connection!
 
   private
     def connect!
-      @bunny ||= create_bunny_connection
-      @bunny.start
+      # If we've already got a bunny object, use it.  This allows people to
+      # specify all kinds of options we don't need to know about (e.g. for ssl).
+      @bunny = @opts[:connection]
+      if @bunny.respond_to?(:call)
+        @bunny = @bunny.call
+      else
+        @bunny ||= create_bunny_connection
+        @bunny.start
+      end
       @channel = @bunny.create_channel
       @exchange = @channel.exchange(@opts[:exchange], **@opts[:exchange_options])
     end

lib/sneakers/queue.rb

@@ -19,8 +19,12 @@ def subscribe(worker)
     # If we've already got a bunny object, use it.  This allows people to
     # specify all kinds of options we don't need to know about (e.g. for ssl).
     @bunny = @opts[:connection]
-    @bunny ||= create_bunny_connection
-    @bunny.start
+    if @bunny.respond_to?(:call)
+      @bunny = @bunny.call
+    else
+      @bunny ||= create_bunny_connection
+      @bunny.start
+    end
 
     @channel = @bunny.create_channel
     @channel.prefetch(@opts[:prefetch])

spec/sneakers/publisher_spec.rb

@@ -104,7 +104,7 @@
         exchange = Object.new
         existing_session = Bunny.new
 
-        mock(existing_session).start
+        mock(existing_session).start { existing_session }
         mock(existing_session).create_channel { channel }
 
         mock(channel).exchange('another_exchange', type: :topic, durable: false, :auto_delete => false, arguments: { 'x-arg' => 'value' }) do
@@ -124,15 +124,15 @@
         @existing_session = existing_session
       end
 
-      it 'can handle an existing connection that is offline' do
+      it 'can handle an existing connection object' do
         p = Sneakers::Publisher.new
         p.publish('test msg', my_vars)
         _(p.instance_variable_get(:@bunny)).must_equal @existing_session
       end
 
-      it 'can handle an existing connection that is online' do
-        mock(@existing_session).connected? { true }
-        p = Sneakers::Publisher.new
+      it 'can handle an existing connection function' do
+        @existing_session.start
+        p = Sneakers::Publisher.new(connection: ->() { @existing_session })
         p.publish('test msg', my_vars)
         _(p.instance_variable_get(:@bunny)).must_equal @existing_session
       end

spec/sneakers/queue_spec.rb

@@ -150,19 +150,24 @@
                                :type => :direct,
                                :durable => true,
                                :arguments => { 'x-arg' => 'value' }){ @mkex }
-
-        queue_name = 'foo'
-        mock(@mkchan).queue(queue_name, :durable => true) { @mkqueue }
-        mock(@mkqueue).bind(@mkex, :routing_key => queue_name)
+        mock(@mkchan).queue('foo', :durable => true) { @mkqueue }
+        mock(@mkqueue).bind(@mkex, :routing_key => 'foo')
         mock(@mkqueue).subscribe(:block => false, :manual_ack => true)
-
-        my_vars = queue_vars.merge(:connection => @external_connection)
-        @q = Sneakers::Queue.new(queue_name, my_vars)
       end
 
       it 'uses that object' do
-        @q.subscribe(@mkworker)
-        _(@q.instance_variable_get(:@bunny)).must_equal @external_connection
+        q = Sneakers::Queue.new('foo',
+                                queue_vars.merge(:connection => @external_connection))
+        q.subscribe(@mkworker)
+        _(q.instance_variable_get(:@bunny)).must_equal @external_connection
+      end
+
+      it 'uses that function' do
+        @external_connection.start
+        q = Sneakers::Queue.new('foo',
+                                queue_vars.merge(:connection => ->() { @external_connection }))
+        q.subscribe(@mkworker)
+        _(q.instance_variable_get(:@bunny)).must_equal @external_connection
       end
     end
   end