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

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

Constant Summary collapse

BYTES_TO_READ = 
8192
SHUTDOWN_TIMEOUT = 
5

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(watcher, path:, read_lines_limit:, read_bytes_limit_per_second:, max_line_size: nil, log:, open_on_every_update:, from_encoding: nil, encoding: nil, metrics:, &receive_lines) ⇒ IOHandler

Returns a new instance of IOHandler.



1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
 
# File 'lib/fluent/plugin/in_tail.rb', line 1113

def initialize(watcher, path:, read_lines_limit:, read_bytes_limit_per_second:, max_line_size: nil, log:, open_on_every_update:, from_encoding: nil, encoding: nil, metrics:, &receive_lines)
  @watcher = watcher
  @path = path
  @read_lines_limit = read_lines_limit
  @read_bytes_limit_per_second = read_bytes_limit_per_second
  @receive_lines = receive_lines
  @open_on_every_update = open_on_every_update
  @fifo = FIFO.new(from_encoding || Encoding::ASCII_8BIT, encoding || Encoding::ASCII_8BIT, log, max_line_size)
  @iobuf = ''.force_encoding('ASCII-8BIT')
  @lines = []
  @io = nil
  @notify_mutex = Mutex.new
  @log = log
  @start_reading_time = nil
  @number_bytes_read = 0
  @shutdown_start_time = nil
  @shutdown_timeout = SHUTDOWN_TIMEOUT
  @shutdown_mutex = Mutex.new
  @eof = false
  @metrics = metrics

  @log.info "following tail of #{@path}"
end

Instance Attribute Details

#shutdown_timeout ⇒ Object

Returns the value of attribute shutdown_timeout.



1111
1112
1113
 
# File 'lib/fluent/plugin/in_tail.rb', line 1111

def shutdown_timeout
  @shutdown_timeout
end

Instance Method Details

#close ⇒ Object 



1152
1153
1154
1155
1156
1157
1158
 
# File 'lib/fluent/plugin/in_tail.rb', line 1152

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

#eof? ⇒ Boolean

Returns:

  • (Boolean) 


1164
1165
1166
 
# File 'lib/fluent/plugin/in_tail.rb', line 1164

def eof?
  @eof
end

#group_watcher ⇒ Object 



1137
1138
1139
 
# File 'lib/fluent/plugin/in_tail.rb', line 1137

def group_watcher
  @watcher.group_watcher
end

#on_notify ⇒ Object 



1141
1142
1143
 
# File 'lib/fluent/plugin/in_tail.rb', line 1141

def on_notify
  @notify_mutex.synchronize { handle_notify }
end

#opened? ⇒ Boolean

Returns:

  • (Boolean) 


1160
1161
1162
 
# File 'lib/fluent/plugin/in_tail.rb', line 1160

def opened?
  !!@io
end

#ready_to_shutdown(shutdown_start_time = nil) ⇒ Object 



1145
1146
1147
1148
1149
1150
 
# File 'lib/fluent/plugin/in_tail.rb', line 1145

def ready_to_shutdown(shutdown_start_time = nil)
  @shutdown_mutex.synchronize {
    @shutdown_start_time =
      shutdown_start_time || Fluent::Clock.now
  }
end