# File lib/fluent/command/cat.rb, line 115 def initialize(tag, connector) @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 super() end
# File lib/fluent/command/cat.rb, line 171 def close @socket.close @socket = nil end
# File lib/fluent/command/cat.rb, line 152 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
# File lib/fluent/command/cat.rb, line 182 def shutdown @timer.shutdown end
# File lib/fluent/command/cat.rb, line 176 def start @timer = TimerThread.new(self) @timer.start self end
# File lib/fluent/command/cat.rb, line 132 def write(record) if record.class != Hash raise ArgumentError, "Input must be a map (got #{record.class})" end entry = [Time.now.to_i, 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
# File lib/fluent/command/cat.rb, line 250 def abort_message(time, record) $stdout.puts "!#{time}:#{Yajl.dump(record)}" end
# File lib/fluent/command/cat.rb, line 205 def get_socket unless @socket unless try_connect return nil end end @socket_time = Time.now.to_i return @socket end
# File lib/fluent/command/cat.rb, line 216 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
# File lib/fluent/command/cat.rb, line 187 def write_impl(array) socket = get_socket unless socket return false end begin socket.write [@tag, array].to_msgpack socket.flush rescue $stderr.puts "write failed: #{$!}" close return false end return true end