handle rotations better.

especially in the case where the rotated files are discoverable - in
this case we were not the moving the state and the sincedb record
correctly.
Next - abstract the *nix and Windows stat calls and structure into
two classes with the same API.

Author: Guy Boertje

Diff for: lib/filewatch/bootstrap.rb

@@ -11,7 +11,7 @@ module FileWatch
   # this is used in the read loop e.g.
   # @opts[:file_chunk_count].times do
   # where file_chunk_count defaults to this constant
-  FIXNUM_MAX = (2**(0.size * 8 - 2) - 1)
+  MAX_ITERATIONS = (2**(0.size * 8 - 2) - 2) / 32768
 
   require_relative "helper"
 

Diff for: lib/filewatch/discoverer.rb

@@ -37,7 +37,7 @@ def can_exclude?(watched_file, new_discovery)
       @exclude.each do |pattern|
         if watched_file.pathname.fnmatch?(pattern)
           if new_discovery
-            logger.debug("Discoverer can_exclude?: #{watched_file.path}: skipping " +
+            logger.trace("Discoverer can_exclude?: #{watched_file.path}: skipping " +
               "because it matches exclude #{pattern}")
           end
           watched_file.unwatch
@@ -49,41 +49,42 @@ def can_exclude?(watched_file, new_discovery)
 
     def discover_files(path)
       fileset = Dir.glob(path).select{|f| File.file?(f) && !File.symlink?(f)}
-      logger.debug("Discoverer found files, count: #{fileset.size}")
+      logger.trace("Discoverer found files, count: #{fileset.size}")
+      rotated_paths = []
       fileset.each do |file|
-        logger.debug("Discoverer found file, path: #{file}")
+        logger.trace("Discoverer found file, path: #{file}")
         pathname = Pathname.new(file)
         new_discovery = false
         watched_file = @watched_files_collection.watched_file_by_path(file)
         if watched_file.nil?
           new_discovery = true
           watched_file = WatchedFile.new(pathname, pathname.stat, @settings)
-          logger.debug("Discoverer new discovery: #{watched_file.path}, inode: #{watched_file.sincedb_key.inode}, discover_files: #{path} (exclude is #{@exclude.inspect})")
+          logger.trace("Discoverer new discovery: #{watched_file.path}, inode: #{watched_file.sincedb_key.inode}")
         end
         # if it already unwatched or its excluded then we can skip
         next if watched_file.unwatched? || can_exclude?(watched_file, new_discovery)
 
         if new_discovery
-          if watched_file.file_ignorable?
-            logger.debug("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
-            # on discovery ignorable watched_files are put into the ignored state and that
-            # updates the size from the internal stat
-            # so the existing contents are not read.
-            # because, normally, a newly discovered file will
-            # have a watched_file size of zero
-            # they are still added to the collection so we know they are there for the next periodic discovery
-            watched_file.ignore
-          end
-
-          # now add the discovered file to the watched_files collection and adjust the sincedb collections
-          @watched_files_collection.add(watched_file)
           # initially when the sincedb collection is filled with records from the persistence file
           # each value is not associated with a watched file
           # a sincedb_value can be:
           #   unassociated
           #   associated with this watched_file
           #   associated with a different watched_file
-          @sincedb_collection.associate(watched_file)
+          if @sincedb_collection.associate(watched_file)
+            if watched_file.file_ignorable?
+              logger.trace("Discoverer discover_files: #{file}: skipping because it was last modified more than #{@settings.ignore_older} seconds ago")
+              # on discovery ignorable watched_files are put into the ignored state and that
+              # updates the size from the internal stat
+              # so the existing contents are not read.
+              # because, normally, a newly discovered file will
+              # have a watched_file size of zero
+              # they are still added to the collection so we know they are there for the next periodic discovery
+              watched_file.ignore
+            end
+            # now add the discovered file to the watched_files collection and adjust the sincedb collections
+            @watched_files_collection.add(watched_file)
+          end
         end
         # at this point the watched file is created, is in the db but not yet opened or being processed
       end

Diff for: lib/filewatch/observing_base.rb

@@ -44,7 +44,8 @@ def initialize(opts={})
         :exclude => [],
         :start_new_files_at => :end,
         :delimiter => "\n",
-        :file_chunk_count => FIXNUM_MAX,
+        :file_chunk_count => MAX_ITERATIONS,
+        :file_chunk_size => FILE_READ_SIZE,
         :file_sort_by => "last_modified",
         :file_sort_direction => "asc",
       }.merge(opts)

Diff for: lib/filewatch/read_mode/handlers/base.rb

@@ -14,7 +14,7 @@ def initialize(sincedb_collection, observer, settings)
     end
 
     def handle(watched_file)
-      logger.debug("handling: #{watched_file.path}")
+      logger.trace("handling: #{watched_file.path}")
       unless watched_file.has_listener?
         watched_file.set_listener(@observer)
       end
@@ -29,7 +29,7 @@ def handle_specifically(watched_file)
 
     def open_file(watched_file)
       return true if watched_file.file_open?
-      logger.debug("opening #{watched_file.path}")
+      logger.trace("opening #{watched_file.path}")
       begin
         watched_file.open
       rescue
@@ -41,7 +41,7 @@ def open_file(watched_file)
           logger.warn("failed to open #{watched_file.path}: #{$!.inspect}, #{$!.backtrace.take(3)}")
           watched_file.last_open_warning_at = now
         else
-          logger.debug("suppressed warning for `failed to open` #{watched_file.path}: #{$!.inspect}")
+          logger.trace("suppressed warning for `failed to open` #{watched_file.path}: #{$!.inspect}")
         end
         watched_file.watch # set it back to watch so we can try it again
       end
@@ -60,21 +60,34 @@ def add_or_update_sincedb_collection(watched_file)
       elsif sincedb_value.watched_file == watched_file
         update_existing_sincedb_collection_value(watched_file, sincedb_value)
       else
-        logger.warn? && logger.warn("mismatch on sincedb_value.watched_file, this should have been handled by Discoverer")
+        msg = "add_or_update_sincedb_collection: the found sincedb_value has a watched_file - this is a rename, switching inode to this watched file"
+        logger.trace(msg)
+        existing_watched_file = sincedb_value.watched_file
+        if existing_watched_file.nil?
+          sincedb_value.set_watched_file(watched_file)
+          logger.trace("add_or_update_sincedb_collection: switching as new file")
+          watched_file.rotate_as_file
+          watched_file.update_bytes_read(sincedb_value.position)
+        else
+          sincedb_value.set_watched_file(watched_file)
+          logger.trace("add_or_update_sincedb_collection: switching from...", "watched_file details" => watched_file.details)
+          watched_file.rotate_from(existing_watched_file)
+        end
+
       end
       watched_file.initial_completed
     end
 
     def update_existing_sincedb_collection_value(watched_file, sincedb_value)
-      logger.debug("update_existing_sincedb_collection_value: #{watched_file.path}, last value #{sincedb_value.position}, cur size #{watched_file.last_stat_size}")
+      logger.trace("update_existing_sincedb_collection_value: #{watched_file.path}, last value #{sincedb_value.position}, cur size #{watched_file.last_stat_size}")
       # sincedb_value is the source of truth
       watched_file.update_bytes_read(sincedb_value.position)
     end
 
     def add_new_value_sincedb_collection(watched_file)
       sincedb_value = SincedbValue.new(0)
       sincedb_value.set_watched_file(watched_file)
-      logger.debug("add_new_value_sincedb_collection: #{watched_file.path}", "position" => sincedb_value.position)
+      logger.trace("add_new_value_sincedb_collection: #{watched_file.path}", "position" => sincedb_value.position)
       sincedb_collection.set(watched_file.sincedb_key, sincedb_value)
     end
   end

Diff for: lib/filewatch/read_mode/handlers/read_file.rb

@@ -9,7 +9,7 @@ def handle_specifically(watched_file)
         watched_file.read_loop_count.times do
           begin
             result = watched_file.read_extract_lines # expect BufferExtractResult
-            logger.info(result.warning, result.additional) unless result.warning.empty?
+            logger.trace(result.warning, result.additional) unless result.warning.empty?
             changed = true
             result.lines.each do |line|
               watched_file.listener.accept(line)
@@ -22,8 +22,9 @@ def handle_specifically(watched_file)
             watched_file.listener.accept(line) unless line.empty?
             watched_file.listener.eof
             watched_file.file_close
-            # unset_watched_file will set sincedb_value.position to be watched_file.bytes_read
-            sincedb_collection.unset_watched_file(watched_file)
+            key = watched_file.sincedb_key
+            sincedb_collection.reading_completed(key)
+            sincedb_collection.clear_watched_file(key)
             watched_file.listener.deleted
             watched_file.unwatch
             break

Diff for: lib/filewatch/read_mode/handlers/read_zip_file.rb

@@ -17,6 +17,9 @@ def handle_specifically(watched_file)
       # should we track lines read in the sincedb and
       # fast forward through the lines until we reach unseen content?
       # meaning that we can quit in the middle of a zip file
+      key = watched_file.sincedb_key
+            sincedb_collection.reading_completed(key)
+
       begin
         file_stream = FileInputStream.new(watched_file.path)
         gzip_stream = GZIPInputStream.new(file_stream)
@@ -30,7 +33,7 @@ def handle_specifically(watched_file)
         logger.error("Cannot decompress the gzip file at path: #{watched_file.path}")
         watched_file.listener.error
       else
-        sincedb_collection.store_last_read(watched_file.sincedb_key, watched_file.last_stat_size)
+        sincedb_collection.store_last_read(key, watched_file.last_stat_size)
         sincedb_collection.request_disk_flush
         watched_file.listener.deleted
         watched_file.unwatch
@@ -41,7 +44,7 @@ def handle_specifically(watched_file)
         close_and_ignore_ioexception(gzip_stream) unless gzip_stream.nil?
         close_and_ignore_ioexception(file_stream) unless file_stream.nil?
       end
-      sincedb_collection.unset_watched_file(watched_file)
+      sincedb_collection.clear_watched_file(key)
     end
 
     private

Diff for: lib/filewatch/read_mode/processor.rb

@@ -46,7 +46,7 @@ def process_ignored(watched_files)
     end
 
     def process_watched(watched_files)
-      logger.debug("Watched processing")
+      logger.trace("Watched processing")
       # Handles watched_files in the watched state.
       # for a slice of them:
       #   move to the active state
@@ -79,7 +79,7 @@ def process_watched(watched_files)
     end
 
     def process_active(watched_files)
-      logger.debug("Active processing")
+      logger.trace("Active processing")
       # Handles watched_files in the active state.
       watched_files.select {|wf| wf.active? }.each do |watched_file|
         path = watched_file.path
@@ -107,7 +107,7 @@ def common_deleted_reaction(watched_file, action)
       # file has gone away or we can't read it anymore.
       watched_file.unwatch
       deletable_filepaths << watched_file.path
-      logger.debug("#{action} - stat failed: #{watched_file.path}, removing from collection")
+      logger.trace("#{action} - stat failed: #{watched_file.path}, removing from collection")
     end
 
     def common_error_reaction(path, error, action)

Diff for: lib/filewatch/settings.rb

@@ -22,7 +22,7 @@ def initialize
         :delimiter => "\n",
         :file_chunk_size => FILE_READ_SIZE,
         :max_active => 4095,
-        :file_chunk_count => FIXNUM_MAX,
+        :file_chunk_count => MAX_ITERATIONS,
         :sincedb_clean_after => 14,
         :exclude => [],
         :stat_interval => 1,