mirror of
https://github.com/meineerde/rackstash.git
synced 2025-10-17 14:01:01 +00:00
Write the raw buffer to the event on flush and use #to_h instead of #to_event to create the event
With that, we can lazy-transform the Buffer to the event hash. B using the common `.to_h` protocol, we can also support various other objects here instead of just Buffers (including actual raw Hashes).
This commit is contained in:
parent
0d74a23b22
commit
88d50afbf9
@ -222,18 +222,15 @@ module Rackstash
|
|||||||
@fields ||= Rackstash::Fields::Hash.new(forbidden_keys: FORBIDDEN_FIELDS)
|
@fields ||= Rackstash::Fields::Hash.new(forbidden_keys: FORBIDDEN_FIELDS)
|
||||||
end
|
end
|
||||||
|
|
||||||
# Flush the current buffer to the {#flows} if it is pending.
|
# Flush the current buffer to the {#flows} if it is pending. All data in the
|
||||||
#
|
# buffer will be preserved. You might call {#clear} to start anew.
|
||||||
# After the flush, the existing buffer should not be used anymore. You
|
|
||||||
# should either call {#clear} to remove all volatile data or create a new
|
|
||||||
# buffer instance instead.
|
|
||||||
#
|
#
|
||||||
# @return [self,nil] returns `self` if the buffer was flushed, `nil`
|
# @return [self,nil] returns `self` if the buffer was flushed, `nil`
|
||||||
# otherwise
|
# otherwise
|
||||||
def flush
|
def flush
|
||||||
return unless pending?
|
return unless pending?
|
||||||
|
|
||||||
@flows.write(self.to_event)
|
@flows.write(self)
|
||||||
self
|
self
|
||||||
end
|
end
|
||||||
|
|
||||||
@ -347,7 +344,7 @@ module Rackstash
|
|||||||
# All hashes (including nested hashes) use `String` keys.
|
# All hashes (including nested hashes) use `String` keys.
|
||||||
#
|
#
|
||||||
# @return [Hash] the event expected by the event {Filter}s.
|
# @return [Hash] the event expected by the event {Filter}s.
|
||||||
def to_event
|
def to_h
|
||||||
event = fields.to_h
|
event = fields.to_h
|
||||||
event[FIELD_TAGS] = tags.to_a
|
event[FIELD_TAGS] = tags.to_a
|
||||||
event[FIELD_MESSAGE] = messages
|
event[FIELD_MESSAGE] = messages
|
||||||
|
|||||||
@ -179,6 +179,7 @@ module Rackstash
|
|||||||
flows = to_a
|
flows = to_a
|
||||||
flows_size = flows.size
|
flows_size = flows.size
|
||||||
|
|
||||||
|
event = event.to_h
|
||||||
flows.each_with_index do |flow, index|
|
flows.each_with_index do |flow, index|
|
||||||
# If we have more than one flow, we provide a fresh copy of the event
|
# If we have more than one flow, we provide a fresh copy of the event
|
||||||
# to each flow. The flow's filters and encoder can then mutate the event
|
# to each flow. The flow's filters and encoder can then mutate the event
|
||||||
|
|||||||
@ -360,12 +360,11 @@ RSpec.describe Rackstash::Buffer do
|
|||||||
buffer.add_message(message)
|
buffer.add_message(message)
|
||||||
|
|
||||||
# We might call Buffer#flush during the following tests
|
# We might call Buffer#flush during the following tests
|
||||||
allow(buffer).to receive(:to_event).and_return(event)
|
allow(flows).to receive(:write).with(buffer).once
|
||||||
allow(flows).to receive(:write).with(event).once
|
|
||||||
end
|
end
|
||||||
|
|
||||||
it 'flushes the buffer to the flows' do
|
it 'flushes the buffer to the flows' do
|
||||||
expect(flows).to receive(:write).with(event).once
|
expect(flows).to receive(:write).with(buffer).once
|
||||||
buffer.flush
|
buffer.flush
|
||||||
end
|
end
|
||||||
|
|
||||||
@ -563,7 +562,7 @@ RSpec.describe Rackstash::Buffer do
|
|||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
describe '#to_event' do
|
describe '#to_h' do
|
||||||
it 'creates an event hash' do
|
it 'creates an event hash' do
|
||||||
message = double(message: 'Hello World', time: Time.now)
|
message = double(message: 'Hello World', time: Time.now)
|
||||||
allow(message)
|
allow(message)
|
||||||
@ -571,7 +570,7 @@ RSpec.describe Rackstash::Buffer do
|
|||||||
buffer.fields[:foo] = 'bar'
|
buffer.fields[:foo] = 'bar'
|
||||||
buffer.tags << 'some_tag'
|
buffer.tags << 'some_tag'
|
||||||
|
|
||||||
expect(buffer.to_event).to match(
|
expect(buffer.to_h).to match(
|
||||||
'foo' => 'bar',
|
'foo' => 'bar',
|
||||||
'message' => [message],
|
'message' => [message],
|
||||||
'tags' => ['some_tag'],
|
'tags' => ['some_tag'],
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user