mirror of https://github.com/reiseburo/hermann
Add a failing integration test for pumping protobufs through Hermann
This commit is contained in:
parent
7f4de34be0
commit
228358edc1
|
@ -0,0 +1,35 @@
|
||||||
|
##
|
||||||
|
# This file is auto-generated. DO NOT EDIT!
|
||||||
|
#
|
||||||
|
require 'protobuf/message'
|
||||||
|
|
||||||
|
module Hermann
|
||||||
|
|
||||||
|
##
|
||||||
|
# Enum Classes
|
||||||
|
#
|
||||||
|
class States < ::Protobuf::Enum
|
||||||
|
define :FULFILLED, 1
|
||||||
|
define :UNFULFILLED, 2
|
||||||
|
define :PENDING, 3
|
||||||
|
define :REJECTED, 4
|
||||||
|
end
|
||||||
|
|
||||||
|
|
||||||
|
##
|
||||||
|
# Message Classes
|
||||||
|
#
|
||||||
|
class TestEvent < ::Protobuf::Message; end
|
||||||
|
|
||||||
|
|
||||||
|
##
|
||||||
|
# Message Fields
|
||||||
|
#
|
||||||
|
class TestEvent
|
||||||
|
required :string, :name, 1
|
||||||
|
required ::Hermann::States, :state, 2
|
||||||
|
optional :int32, :bogomips, 3
|
||||||
|
end
|
||||||
|
|
||||||
|
end
|
||||||
|
|
|
@ -0,0 +1,16 @@
|
||||||
|
package hermann;
|
||||||
|
|
||||||
|
// Generate Ruby stubs with: protoc --ruby_out=. spec/fixtures/testevent.proto
|
||||||
|
|
||||||
|
enum States {
|
||||||
|
FULFILLED = 1;
|
||||||
|
UNFULFILLED = 2;
|
||||||
|
PENDING = 3;
|
||||||
|
REJECTED = 4;
|
||||||
|
}
|
||||||
|
|
||||||
|
message TestEvent {
|
||||||
|
required string name = 1;
|
||||||
|
required States state = 2;
|
||||||
|
optional int32 bogomips = 3;
|
||||||
|
}
|
|
@ -5,6 +5,9 @@ require 'hermann/consumer'
|
||||||
require 'hermann/discovery/zookeeper'
|
require 'hermann/discovery/zookeeper'
|
||||||
require 'concurrent'
|
require 'concurrent'
|
||||||
|
|
||||||
|
require 'protobuf'
|
||||||
|
require_relative '../fixtures/testevent.pb'
|
||||||
|
|
||||||
describe 'producer' do
|
describe 'producer' do
|
||||||
include_context 'integration test context'
|
include_context 'integration test context'
|
||||||
|
|
||||||
|
@ -15,26 +18,50 @@ describe 'producer' do
|
||||||
end
|
end
|
||||||
let(:consumer_promise) do
|
let(:consumer_promise) do
|
||||||
Concurrent::Promise.execute do
|
Concurrent::Promise.execute do
|
||||||
valid = false
|
value = nil
|
||||||
puts "consuming off `#{topic}`"
|
puts "consuming off `#{topic}`"
|
||||||
consumer.consume(topic) do |dequeued|
|
consumer.consume(topic) do |dequeued|
|
||||||
puts "received the message: #{dequeued}"
|
puts "received the message: #{dequeued.inspect}"
|
||||||
if message == dequeued
|
value = dequeued
|
||||||
consumer.shutdown
|
consumer.shutdown
|
||||||
valid = true
|
|
||||||
end
|
|
||||||
end
|
end
|
||||||
# Return this out of the block
|
# Return this out of the block
|
||||||
next valid
|
next value
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
let(:brokers) do
|
||||||
it 'produces and consumes messages', :type => :integration, :platform => :java do
|
|
||||||
broker_ids = Hermann::Discovery::Zookeeper.new(zookeepers).get_brokers
|
broker_ids = Hermann::Discovery::Zookeeper.new(zookeepers).get_brokers
|
||||||
puts "using ZK discovered brokers: #{broker_ids}"
|
puts "using ZK discovered brokers: #{broker_ids}"
|
||||||
producer = Hermann::Producer.new(nil, broker_ids)
|
broker_ids
|
||||||
|
end
|
||||||
|
let(:producer) { Hermann::Producer.new(nil, brokers) }
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
it 'produces and consumes messages', :type => :integration, :platform => :java do
|
||||||
producer.push(message, :topic => topic).value!(timeout)
|
producer.push(message, :topic => topic).value!(timeout)
|
||||||
expect(consumer_promise.value!(timeout)).to be true
|
expect(consumer_promise.value!(timeout)).to eql(message)
|
||||||
|
end
|
||||||
|
|
||||||
|
|
||||||
|
context 'with binary data', :type => :integration, :platform => :java do
|
||||||
|
let(:event) do
|
||||||
|
Hermann::TestEvent.new(:name => 'rspec',
|
||||||
|
:state => 3,
|
||||||
|
:bogomips => 9001)
|
||||||
|
end
|
||||||
|
|
||||||
|
let(:message) { event.encode }
|
||||||
|
|
||||||
|
it 'should be a thing' do
|
||||||
|
producer.push(message, :topic => topic).value!(timeout)
|
||||||
|
dequeued = consumer_promise.value!(timeout)
|
||||||
|
expect(dequeued).to eql(message)
|
||||||
|
|
||||||
|
expect {
|
||||||
|
Hermann::TestEvent.decode(dequeued)
|
||||||
|
}.not_to raise_error
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
after :each do
|
after :each do
|
||||||
|
|
Loading…
Reference in New Issue