Class: FileWatch::Watch

Inherits:
Object
  • Object
show all
Includes:
LogStash::Util::Loggable
Defined in:
lib/filewatch/watch.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(discoverer, processor, settings) ⇒ Watch

Returns a new instance of Watch.



12
13
14
15
16
17
18
19
20
21
22
23
 
# File 'lib/filewatch/watch.rb', line 12

def initialize(discoverer, processor, settings)
  @discoverer = discoverer
  @watched_files_collection = discoverer.watched_files_collection
  @settings = settings

  # we need to be threadsafe about the quit mutation
  @quit = Concurrent::AtomicBoolean.new(false)
  @lastwarn_max_files = 0

  @processor = processor
  @processor.add_watch(self)
end

Instance Attribute Details

#discoverer ⇒ Object (readonly)

Returns the value of attribute discoverer.



10
11
12
 
# File 'lib/filewatch/watch.rb', line 10

def discoverer
  @discoverer
end

#lastwarn_max_files ⇒ Object

Returns the value of attribute lastwarn_max_files.



9
10
11
 
# File 'lib/filewatch/watch.rb', line 9

def lastwarn_max_files
  @lastwarn_max_files
end

#processor ⇒ Object (readonly)

Returns the value of attribute processor.



10
11
12
 
# File 'lib/filewatch/watch.rb', line 10

def processor
  @processor
end

#watched_files_collection ⇒ Object (readonly)

Returns the value of attribute watched_files_collection.



10
11
12
 
# File 'lib/filewatch/watch.rb', line 10

def watched_files_collection
  @watched_files_collection
end

Instance Method Details

#discover ⇒ Object 



31
32
33
34
35
 
# File 'lib/filewatch/watch.rb', line 31

def discover
  @discoverer.discover
  # don't return whatever @discoverer.discover returns
  return true
end

#iterate_on_state ⇒ Object

Read mode processor will handle watched_files in the closed, ignored, watched and active state differently from Tail mode - see the ReadMode::Processor and TailMode::Processor



65
66
67
68
69
70
71
72
73
74
 
# File 'lib/filewatch/watch.rb', line 65

def iterate_on_state
  return if @watched_files_collection.empty?
  begin
    # creates this snapshot of watched_file values just once
    watched_files = @watched_files_collection.values
    @processor.process_all_states(watched_files)
  ensure
    @watched_files_collection.remove_paths(@processor.clear_deletable_paths)
  end
end

#quit ⇒ Object 



76
77
78
 
# File 'lib/filewatch/watch.rb', line 76

def quit
  @quit.make_true
end

#quit? ⇒ Boolean

Returns:

  • (Boolean) 


80
81
82
 
# File 'lib/filewatch/watch.rb', line 80

def quit?
  @quit.true? || (@settings.exit_after_read && @watched_files_collection.empty?)
end

#subscribe(observer, sincedb_collection) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/filewatch/watch.rb', line 37

def subscribe(observer, sincedb_collection)
  @processor.initialize_handlers(sincedb_collection, observer)

  glob = 0
  interval = @settings.discover_interval
  reset_quit
  until quit?
    iterate_on_state
    # Don't discover new files when files to read are known at the beginning
    break if quit?
    sincedb_collection.write_if_requested
    glob += 1
    if glob == interval && !@settings.exit_after_read
      discover
      glob = 0
    end
    break if quit?
    # NOTE: maybe the plugin should validate stat_interval <= sincedb_write_interval <= sincedb_clean_after
    sleep(@settings.stat_interval)
    # we need to check potential expired keys (sincedb_clean_after) periodically
    sincedb_collection.flush_at_interval
  end
  sincedb_collection.write_if_requested # does nothing if no requests to write were lodged.
  @watched_files_collection.close_all
end

#watch(path) ⇒ Object 



25
26
27
28
29
 
# File 'lib/filewatch/watch.rb', line 25

def watch(path)
  @discoverer.add_path(path)
  # don't return whatever @discoverer.add_path returns
  return true
end