Method: FileWatch::Watch#each

Defined in:

lib/filewatch/watch.rb

#each(&block) ⇒ Object

Calls &block with params [event_type, path] event_type can be one of:

:create_initial - initially present file (so start at end for tail)
:create - file is created (new file after initial globs, start at 0)
:modify - file is modified (size increases)
:delete - file is deleted
:timeout - file is closable
:unignore - file was ignored, but since then it received new content

# File 'lib/filewatch/watch.rb', line 117

def each(&block)
  synchronized do
    return if @files.empty?
    begin
      file_deletable = []
      # creates this array just once
      watched_files = @files.values

      # look at the closed to see if its changed
      watched_files.select {|wf| wf.closed? }.each do |watched_file|
        path = watched_file.path
        break if quit?
        begin
          stat = watched_file.restat
          if watched_file.size_changed? || watched_file.inode_changed?(inode(path,stat))
            # if the closed file changed, move it to the watched state
            # not to active state because we want to use MAX_OPEN_FILES throttling.
            watched_file.watch
          end
        rescue Errno::ENOENT
          # file has gone away or we can't read it anymore.
          file_deletable << path
          @logger.debug? && @logger.debug("each: closed?: stat failed: #{path}: (#{$!}), deleting from @files")
        rescue => e
          @logger.error("each: closed?: #{path}: (#{e.inspect})")
        end
      end
      return if quit?

      # look at the ignored to see if its changed
      watched_files.select {|wf| wf.ignored? }.each do |watched_file|
        path = watched_file.path
        break if quit?
        begin
          stat = watched_file.restat
          if watched_file.size_changed? || watched_file.inode_changed?(inode(path,stat))
            # if the ignored file changed, move it to the watched state
            # not to active state because we want to use MAX_OPEN_FILES throttling.
            # this file has not been yielded to the block yet
            # but we must have the tail to start from the end, so when the file
            # was first ignored we updated the bytes_read to the stat.size at that time.
            # by adding this to the sincedb so that the subsequent modify
            # event can detect the change
            watched_file.watch
            yield(:unignore, watched_file)
          end
        rescue Errno::ENOENT
          # file has gone away or we can't read it anymore.
          file_deletable << path
          @logger.debug? && @logger.debug("each: ignored: stat failed: #{path}: (#{$!}), deleting from @files")
        rescue => e
          @logger.error("each: ignored?: #{path}: (#{e.inspect})")
        end
      end

      return if quit?

      # Send any creates.
      if (to_take = @max_active - watched_files.count{|wf| wf.active?}) > 0
        watched_files.select {|wf| wf.watched? }.take(to_take).each do |watched_file|
          break if quit?
          path = watched_file.path
          begin
            stat = watched_file.restat
            watched_file.activate
            # don't do create again
            next if watched_file.state_history_any?(:closed, :ignored)
            # if the file can't be opened during the yield
            # its state is set back to watched
            sym = watched_file.initial? ? :create_initial : :create
            yield(sym, watched_file)
          rescue Errno::ENOENT
            # file has gone away or we can't read it anymore.
            file_deletable << path
            watched_file.unwatch
            yield(:delete, watched_file)
            next
            @logger.debug? && @logger.debug("each: closed: stat failed: #{path}: (#{$!}), deleting from @files")
          rescue => e
            @logger.error("each: watched?: #{path}: (#{e.inspect})")
          end
        end
      else
        now = Time.now.to_i
        if (now - @lastwarn_max_files) > MAX_FILES_WARN_INTERVAL
          waiting = @files.size - @max_active
          specific = if @close_older.nil?
           ", try setting close_older. There are #{waiting} unopened files"
          else
            ", files yet to open: #{waiting}"
          end
          @logger.warn(@max_warn_msg + specific)
          @lastwarn_max_files = now
        end
      end

      return if quit?

      # wf.active means the actual files were opened
      # and have been read once - unless they were empty at the time
      watched_files.select {|wf| wf.active? }.each do |watched_file|
        path = watched_file.path
        break if quit?
        begin
          stat = watched_file.restat
        rescue Errno::ENOENT
          # file has gone away or we can't read it anymore.
          file_deletable << path
          @logger.debug? && @logger.debug("each: active: stat failed: #{path}: (#{$!}), deleting from @files")
          watched_file.unwatch
          yield(:delete, watched_file)
          next
        rescue => e
          @logger.error("each: active?: #{path}: (#{e.inspect})")
          next
        end

        if watched_file.file_closable?
          @logger.debug? && @logger.debug("each: active: file expired: #{path}")
          yield(:timeout, watched_file)
          watched_file.close
          next
        end

        _inode = inode(path,stat)
        read_thus_far = watched_file.bytes_read
        # we don't update the size here, its updated when we actually read
        if watched_file.inode_changed?(_inode)
          @logger.debug? && @logger.debug("each: new inode: #{path}: old inode was #{watched_file.inode.inspect}, new is #{_inode.inspect}")
          watched_file.update_inode(_inode)
          yield(:delete, watched_file)
          yield(:create, watched_file)
        elsif stat.size < read_thus_far
          @logger.debug? && @logger.debug("each: file rolled: #{path}: new size is #{stat.size}, old size #{read_thus_far}")
          yield(:delete, watched_file)
          yield(:create, watched_file)
        elsif stat.size > read_thus_far
          @logger.debug? && @logger.debug("each: file grew: #{path}: old size #{read_thus_far}, new size #{stat.size}")
          yield(:modify, watched_file)
        end
      end
    ensure
      file_deletable.each {|f| @files.delete(f)}
    end
  end
end