class Writer
Public Class Methods
new(tag, connector, time_as_integer: false)
click to toggle source
Calls superclass method
# File lib/fluent/command/cat.rb, line 131 def initialize(tag, connector, time_as_integer: false) @tag = tag @connector = connector @socket = false @socket_time = Time.now.to_i @socket_ttl = 10 # TODO @error_history = [] @pending = [] @pending_limit = 1024 # TODO @retry_wait = 1 @retry_limit = 5 # TODO @time_as_integer = time_as_integer super() end
Public Instance Methods
close()
click to toggle source
# File lib/fluent/command/cat.rb, line 190 def close @socket.close @socket = nil end
on_timer()
click to toggle source
# File lib/fluent/command/cat.rb, line 171 def on_timer now = Time.now.to_i synchronize { unless @pending.empty? # flush pending records if write_impl(@pending) # write succeeded @pending.clear end end if @socket && @socket_time + @socket_ttl < now # socket is not used @socket_ttl seconds close end } end
shutdown()
click to toggle source
# File lib/fluent/command/cat.rb, line 201 def shutdown @timer.shutdown end
start()
click to toggle source
# File lib/fluent/command/cat.rb, line 195 def start @timer = TimerThread.new(self) @timer.start self end
write(record)
click to toggle source
# File lib/fluent/command/cat.rb, line 149 def write(record) if record.class != Hash raise ArgumentError, "Input must be a map (got #{record.class})" end time = Fluent::EventTime.now time = time.to_i if @time_as_integer entry = [time, record] synchronize { unless write_impl([entry]) # write failed @pending.push(entry) while @pending.size > @pending_limit # exceeds pending limit; trash oldest record time, record = @pending.shift abort_message(time, record) end end } end
Private Instance Methods
abort_message(time, record)
click to toggle source
# File lib/fluent/command/cat.rb, line 270 def abort_message(time, record) $stdout.puts "!#{time}:#{Yajl.dump(record)}" end
get_socket()
click to toggle source
# File lib/fluent/command/cat.rb, line 225 def get_socket unless @socket unless try_connect return nil end end @socket_time = Time.now.to_i return @socket end
try_connect()
click to toggle source
# File lib/fluent/command/cat.rb, line 236 def try_connect now = Time.now.to_i unless @error_history.empty? # wait before re-connecting wait = @retry_wait * (2 ** (@error_history.size-1)) if now <= @socket_time + wait return false end end begin @socket = @connector.call @error_history.clear return true rescue $stderr.puts "connect failed: #{$!}" @error_history << $! @socket_time = now if @retry_limit < @error_history.size # abort all pending records @pending.each {|(time, record)| abort_message(time, record) } @pending.clear @error_history.clear end return false end end
write_impl(array)
click to toggle source
# File lib/fluent/command/cat.rb, line 206 def write_impl(array) socket = get_socket unless socket return false end begin packer = Fluent::MessagePackFactory.packer socket.write packer.pack([@tag, array]) socket.flush rescue $stderr.puts "write failed: #{$!}" close return false end return true end