forked from codekitchen/logstash-input-kinesis
-
Notifications
You must be signed in to change notification settings - Fork 0
/
worker_spec.rb
58 lines (49 loc) · 1.91 KB
/
worker_spec.rb
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
require "logstash/plugin"
require "logstash/inputs/kinesis"
require "logstash/codecs/json"
require "json"
RSpec.describe "LogStash::Inputs::Kinesis::Worker" do
subject!(:worker) { LogStash::Inputs::Kinesis::Worker.new(codec, queue, decorator, checkpoint_interval) }
let(:codec) { LogStash::Codecs::JSON.new() }
let(:queue) { Queue.new }
let(:decorator) { proc { |x| x["decorated"] = true; x } }
let(:checkpoint_interval) { 120 }
let(:checkpointer) { double('checkpointer', checkpoint: nil) }
let(:shard_id) { "xyz" }
it "honors the initialize java interface method contract" do
expect { worker.initialize(shard_id) }.to_not raise_error
end
def record(hash = { "message" => "test" })
encoder = java.nio.charset::Charset.forName("UTF-8").newEncoder()
data = encoder.encode(java.nio.CharBuffer.wrap(JSON.generate(hash)))
double(getData: data)
end
let(:record1) { record(id: "record1", message: "test1") }
let(:record2) { record(id: "record2", message: "test2") }
context "initialized" do
before do
worker.initialize(shard_id)
end
describe "#processRecords" do
it "decodes and queues each record with decoration" do
worker.processRecords([record1, record2], checkpointer)
m1 = queue.pop
m2 = queue.pop
expect(m1).to be_kind_of(LogStash::Event)
expect(m2).to be_kind_of(LogStash::Event)
expect(m1['id']).to eq("record1")
expect(m1['message']).to eq("test1")
expect(m1['decorated']).to eq(true)
end
it "checkpoints on interval" do
expect(checkpointer).to receive(:checkpoint).once
worker.processRecords([], checkpointer)
# not this time
worker.processRecords([], checkpointer)
allow(Time).to receive(:now).and_return(Time.now + 125)
expect(checkpointer).to receive(:checkpoint).once
worker.processRecords([], checkpointer)
end
end
end
end