Class: Fluent::Plugin::TailInput::TailWatcher::IOHandler

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/plugin/in_tail.rb

Instance Method Summary collapse

Constructor Details

#initialize(watcher, &receive_lines) ⇒ IOHandler

Returns a new instance of IOHandler.



624
625
626
627
628
629
630
631
632
 
# File 'lib/fluent/plugin/in_tail.rb', line 624

def initialize(watcher, &receive_lines)
  @watcher = watcher
  @receive_lines = receive_lines
  @fifo = FIFO.new(@watcher.from_encoding || Encoding::ASCII_8BIT, @watcher.encoding || Encoding::ASCII_8BIT)
  @iobuf = ''.force_encoding('ASCII-8BIT')
  @lines = []
  @io = nil
  @watcher.log.info "following tail of #{@watcher.path}"
end

Instance Method Details

#close ⇒ Object 



668
669
670
671
672
673
 
# File 'lib/fluent/plugin/in_tail.rb', line 668

def close
  if @io && !@io.closed?
    @io.close
    @io = nil
  end
end

#on_notify ⇒ Object 



634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
 
# File 'lib/fluent/plugin/in_tail.rb', line 634

def on_notify
  with_io do |io|
    begin
      read_more = false

      if !io.nil? && @lines.empty?
        begin
          while true
            @fifo << io.readpartial(2048, @iobuf)
            while (line = @fifo.next_line)
              @lines << line
            end
            if @lines.size >= @watcher.read_lines_limit
              # not to use too much memory in case the file is very large
              read_more = true
              break
            end
          end
        rescue EOFError
        end
      end

      unless @lines.empty?
        if @receive_lines.call(@lines)
          @watcher.pe.update_pos(io.pos - @fifo.bytesize)
          @lines.clear
        else
          read_more = false
        end
      end
    end while read_more
  end
end

#open ⇒ Object 



679
680
681
682
683
684
685
 
# File 'lib/fluent/plugin/in_tail.rb', line 679

def open
  io = Fluent::FileWrapper.open(@watcher.path)
  io.seek(@watcher.pe.read_pos + @fifo.bytesize)
  io
rescue Errno::ENOENT
  nil
end

#opened? ⇒ Boolean

Returns:

  • (Boolean) 


675
676
677
 
# File 'lib/fluent/plugin/in_tail.rb', line 675

def opened?
  !!@io
end

#with_io ⇒ Object 



687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
 
# File 'lib/fluent/plugin/in_tail.rb', line 687

def with_io
  begin
    if @watcher.open_on_every_update
      io = open
      begin
        yield io
      ensure
        io.close unless io.nil?
      end
    else
      @io ||= open
      yield @io
    end
  rescue
    @watcher.log.error $!.to_s
    @watcher.log.error_backtrace
    close
  end
end