class Franz::Discover

Discover performs half of file existence detection by expanding globs and keeping track of files known to Franz. Discover requires a deletions Queue to maintain this state, so it’s fairly useless without a Watch.

Attributes

configs[R]
deletions[R]
discover_interval[R]
discoveries[R]
known[R]

Public Class Methods

new(opts={}) click to toggle source

Start a new Discover thread in the background.

@param [Hash] opts options for the discovery @option opts [InputConfig] :input_config shared Franz configuration @option opts [Queue] :discoveries ([]) “output” queue of discovered paths @option opts [Queue] :deletions ([]) “input” queue of deleted paths @option opts [Integer] :discover_interval (5) seconds between discover rounds @option opts [Array<Path>] :known ([]) internal “known” state @option opts [Logger] :logger (Logger.new(STDOUT)) logger to use

# File lib/franz/discover.rb, line 22
def initialize opts={}
  @ic = opts[:input_config] || raise('No input_config specified')

  @discoveries = opts[:discoveries] || []
  @deletions   = opts[:deletions]   || []

  @discover_interval = opts[:discover_interval] || 30
  @known             = opts[:known]             || []
  @logger            = opts[:logger]            || Logger.new(STDOUT)

  @known = Set.new(@known)

  @configs = @ic.configs.map do |config|
    config[:includes] ||= []
    config[:excludes] ||= []
    config
  end

  @statz = opts[:statz] || Franz::Stats.new
  @statz.create :num_discovered, 0
  @statz.create :num_deleted, 0

  @stop = false

  @thread = Thread.new do
    until @stop
      until deletions.empty?
        d = deletions.pop
        @known.delete d
        @statz.inc :num_deleted
        log.debug \
          event: 'discover deleted',
          file: d
      end

      discover.each do |discovery|
        discoveries.push discovery
        @known.add discovery
        @statz.inc :num_discovered
        log.debug \
          event: 'discover discovered',
          file: discovery
      end
      sleep discover_interval
    end
  end

  log.debug \
    event: 'discover started',
    discoveries: discoveries,
    deletions: deletions,
    discover_interval: discover_interval
end

Public Instance Methods

state() click to toggle source

Return the internal “known” state

# File lib/franz/discover.rb, line 88
def state
  return @known.to_a
end
stop() click to toggle source

Stop the Discover thread. Effectively only once.

@return [Array] internal “known” state

# File lib/franz/discover.rb, line 79
def stop
  return state if @stop
  @stop = true
  @thread.kill
  log.debug event: 'discover stopped'
  return state
end

Private Instance Methods

discover() click to toggle source
# File lib/franz/discover.rb, line 97
def discover
  log.debug event: 'discover'
  discovered = []
  configs.each do |config|
    config[:includes].each do |glob|
      Dir[glob].each do |path|
        next if known.include? path
        next if config[:excludes].any? { |exclude|
          File.fnmatch? exclude, File::basename(path), File::FNM_EXTGLOB
        }
        discovered.push path
      end
    end
  end
  return discovered
end
log() click to toggle source
# File lib/franz/discover.rb, line 95
def log ; @logger end