From 7f4de34be00d620497183957f176b39b5692c2ad Mon Sep 17 00:00:00 2001
From: "R. Tyler Croy" <tyler@monkeypox.org>
Date: Thu, 30 Oct 2014 10:19:32 -0700
Subject: [PATCH] Refactor the Hermann::Producer Java-based integration test to
 respect integration.yml

I've also refactored/cleaned it up a bit to make sure it runs properly on my machine
---
 spec/integration/producer_spec.rb | 55 ++++++++++++++++---------------
 1 file changed, 28 insertions(+), 27 deletions(-)

diff --git a/spec/integration/producer_spec.rb b/spec/integration/producer_spec.rb
index 34eab67..e336095 100644
--- a/spec/integration/producer_spec.rb
+++ b/spec/integration/producer_spec.rb
@@ -5,39 +5,40 @@ require 'hermann/consumer'
 require 'hermann/discovery/zookeeper'
 require 'concurrent'
 
+describe 'producer' do
+  include_context 'integration test context'
 
-if Hermann.jruby?
-
-  class ConsumerTest
-    def create_consumer
-      zookeeper = "localhost:2181"
-      groupId   = "group1"
-      topic     = 'test'
-
-      consumer = Hermann::Consumer.new(topic, groupId, zookeeper)
-
-      consumer.consume(topic) do |msg|
-        if msg == 'msg'
+  let(:timeout) { 10 }
+  let(:message)   { 'msg' }
+  let(:consumer) do
+    Hermann::Consumer.new(topic, 'rspec-group', zookeepers)
+  end
+  let(:consumer_promise) do
+    Concurrent::Promise.execute do
+      valid = false
+      puts "consuming off `#{topic}`"
+      consumer.consume(topic) do |dequeued|
+        puts "received the message: #{dequeued}"
+        if message == dequeued
           consumer.shutdown
-          return true
+          valid = true
         end
       end
-      false
+      # Return this out of the block
+      next valid
     end
   end
 
-  describe 'producer' do
-    include_context 'integration test context'
-
-    let(:message)   { 'msg' }
-
-    it 'produces and consumes messages', :type => :integration, :platform => :java do
-      test_consumer = Concurrent::Promise.execute { ConsumerTest.new.create_consumer }
-      broker_ids = Hermann::Discovery::Zookeeper.new(zookeepers).get_brokers
-      producer = Hermann::Producer.new(nil, broker_ids)
-      producer.push(message, :topic => topic).wait(1)
-      expect(test_consumer.value(1)).to be true
-    end
+  it 'produces and consumes messages', :type => :integration, :platform => :java do
+    broker_ids = Hermann::Discovery::Zookeeper.new(zookeepers).get_brokers
+    puts "using ZK discovered brokers: #{broker_ids}"
+    producer = Hermann::Producer.new(nil, broker_ids)
+    producer.push(message, :topic => topic).value!(timeout)
+    expect(consumer_promise.value!(timeout)).to be true
   end
 
-end
\ No newline at end of file
+  after :each do
+    # Make sure we shut down our connection in any case
+    consumer.shutdown
+  end
+end