fix timing issues on integration spec

* use timeout and retry if expected consumers are not there
* refactor rabbitmq stuff a little in integration spec

Author: aiomaster

spec/sneakers/integration_spec.rb

@@ -4,6 +4,7 @@
 require 'fixtures/integration_worker'
 
 require "rabbitmq/http/client"
+require 'timeout'
 
 
 describe "integration" do
@@ -17,23 +18,30 @@ def integration_log(msg)
       puts msg if ENV['INTEGRATION_LOG']
     end
 
+    def rmq_addr
+      @rmq_addr ||= compose_or_localhost("rabbitmq")
+    end
+
+    def admin
+      @admin ||=
+        begin
+          puts "RABBITMQ is at #{rmq_addr}"
+          RabbitMQ::HTTP::Client.new("http://#{rmq_addr}:15672/", username: "guest", password: "guest")
+        rescue
+          fail "Rabbitmq admin seems to not exist? you better be running this on Travis or Docker. proceeding.\n#{$!}"
+        end
+    end
+
     def prepare
       # clean up all integration queues; admin interface must be installed
       # in integration env
-      rmq_addr = compose_or_localhost("rabbitmq")
-      puts "RABBITMQ is at #{rmq_addr}"
-      begin
-        admin = RabbitMQ::HTTP::Client.new("http://#{rmq_addr}:15672/", username: "guest", password: "guest")
-        qs = admin.list_queues
-        qs.each do |q|
-          name = q.name
-          if name.start_with? 'integration_'
-            admin.delete_queue('/', name)
-            integration_log "cleaning up #{name}."
-          end
+      qs = admin.list_queues
+      qs.each do |q|
+        name = q.name
+        if name.start_with? 'integration_'
+          admin.delete_queue('/', name)
+          integration_log "cleaning up #{name}."
         end
-      rescue
-        puts "Rabbitmq admin seems to not exist? you better be running this on Travis or Docker. proceeding.\n#{$!}"
       end
 
       Sneakers.clear!
@@ -85,22 +93,27 @@ def start_worker(w)
       pid
     end
 
-    def any_consumers
-      rmq_addr = compose_or_localhost("rabbitmq")
-      result = false
-      begin
-        admin = RabbitMQ::HTTP::Client.new("http://#{rmq_addr}:15672/", username: "guest", password: "guest")
-        qs = admin.list_queues
-        qs.each do |q|
-          if q.name.start_with? 'integration_'
-            puts "We see #{q.consumers} consumers on #{q.name}"
-            return true if q.consumers > 0
-          end
+    def consumers_count
+      qs = admin.list_queues
+      qs.each do |q|
+        if q.name.start_with? 'integration_'
+          return [q.consumers, q.name]
         end
-        return false
-      rescue
-        puts "Rabbitmq admin seems to not exist? you better be running this on Travis or Docker. proceeding.\n#{$!}"
       end
+      return [0, nil]
+    end
+
+    def assert_any_consumers(consumers_should_be_there, maximum_wait_time = 15)
+      Timeout::timeout(maximum_wait_time) do
+        loop do
+          consumers, queue = consumers_count
+          fail 'no queues so no consumers' if consumers_should_be_there && !queue
+          puts "We see #{consumers} consumers on #{queue}"
+          (consumers_should_be_there == consumers.zero?) ? sleep(1) : return
+        end
+      end
+    rescue Timeout::Error
+      fail "Consumers should #{'not' unless consumers_should_be_there} be here but #{consumers} consumers were after #{maximum_wait_time}s waiting."
     end
 
     it 'should be possible to terminate when queue is full' do
@@ -116,11 +129,10 @@ def any_consumers
       end
 
       pid = start_worker(IntegrationWorker)
-      any_consumers.must_equal true
+      assert_any_consumers true
       integration_log "Killing #{pid} now!"
       Process.kill("TERM", pid)
-      sleep(2)
-      any_consumers.must_equal false
+      assert_any_consumers false
     end
 
     it 'should pull down 100 jobs from a real queue' do
@@ -144,5 +156,3 @@ def any_consumers
 
   end
 end
-
-