Sha256: d50caff94d6c3fa629b5b75c6f08c03ef5997fb201f24bc41d65a3b21ca0c197
Contents?: true
Size: 1.2 KB
Versions: 3
Compression:
Stored size: 1.2 KB
Contents
# encoding: utf-8
require "logstash/codecs/base"
class LogStash::Codecs::CompressSpooler < LogStash::Codecs::Base
config_name 'compress_spooler'
config :spool_size, :validate => :number, :default => 50
config :compress_level, :validate => :number, :default => 6
public
def register
require "msgpack"
require "zlib"
@buffer = []
end
public
def decode(data)
z = Zlib::Inflate.new
data = MessagePack.unpack(z.inflate(data))
z.finish
z.close
data.each do |event|
event = LogStash::Event.new(event)
event["@timestamp"] = Time.at(event["@timestamp"]).utc if event["@timestamp"].is_a? Float
yield event
end
end # def decode
public
def encode(data)
if @buffer.length >= @spool_size
z = Zlib::Deflate.new(@compress_level)
@on_event.call data, z.deflate(MessagePack.pack(@buffer), Zlib::FINISH)
z.close
@buffer.clear
else
data["@timestamp"] = data["@timestamp"].to_f
@buffer << data.to_hash
end
end # def encode
public
def teardown
if !@buffer.nil? and @buffer.length > 0
@on_event.call LogStash::Event.new, @buffer
end
@buffer.clear
end
end # class LogStash::Codecs::CompressSpooler
Version data entries
3 entries across 3 versions & 1 rubygems