Merge all other contributions

lib/filewatch/tail.rb

@@ -35,14 +35,17 @@ def initialize(opts={})
       @watch = FileWatch::Watch.new
       @watch.logger = @logger
       @sincedb = {}
-      @sincedb_last_write = 0
+      @sincedb_last_write = Time.now.to_i
+      @sincedb_write_pending = false
+      @sincedb_writing = false
       @statcache = {}
       @opts = {
         :sincedb_write_interval => 10,
         :stat_interval => 1,
         :discover_interval => 5,
         :exclude => [],
-        :start_new_files_at => :end
+        :start_new_files_at => :end,
+        :follow_only_path => false
       }.merge(opts)
       if [email protected]?(:sincedb_path)
         @opts[:sincedb_path] = File.join(ENV["HOME"], ".sincedb") if ENV.include?("HOME")
@@ -96,6 +99,9 @@ def subscribe(&block)
           @files[path].close
           @files.delete(path)
           @statcache.delete(path)
+        when :noupdate
+          @logger.debug(":noupdate for #{path}, from @files")
+          _sincedb_write_if_pending   # will check to see if sincedb_write requests are pending 
         else
           @logger.warn("unknown event type #{event} for #{path}")
         end
@@ -128,13 +134,21 @@ def _open_file(path, event)
 
       stat = File::Stat.new(path)
 
-      if @iswindows
-        fileId = Winhelper.GetWindowsUniqueFileIdentifier(path)
-        inode = [fileId, stat.dev_major, stat.dev_minor]
+      if @opts[:follow_only_path]
+        # In cases where files are rsynced to the consuming server, inodes will change when 
+        # updated files overwrite original ones, resulting in inode changes.  In order to 
+        # avoid having the sincedb.member check from failing in this scenario, we'll 
+        # construct the inode key using the path which will be 'stable'
+        inode = [path, stat.dev_major, stat.dev_minor]
       else
-        inode = [stat.ino.to_s, stat.dev_major, stat.dev_minor]
+        if @iswindows
+          fileId = Winhelper.GetWindowsUniqueFileIdentifier(path)
+          inode = [fileId, stat.dev_major, stat.dev_minor]
+        else
+          inode = [stat.ino.to_s, stat.dev_major, stat.dev_minor]
         end
-
+      end
+
       @statcache[path] = inode
 
       if @sincedb.member?(inode)
@@ -186,20 +200,14 @@ def _read_file(path, &block)
       end
 
       if changed
-        now = Time.now.to_i
-        delta = now - @sincedb_last_write
-        if delta >= @opts[:sincedb_write_interval]
-          @logger.debug("writing sincedb (delta since last write = #{delta})")
-          _sincedb_write
-          @sincedb_last_write = now
-        end
+        _sincedb_write
       end
     end # def _read_file
 
     public
     def sincedb_write(reason=nil)
       @logger.debug("caller requested sincedb write (#{reason})")
-      _sincedb_write
+      _sincedb_write(true)  # since this is an external request, force the write
     end
 
     private
@@ -214,26 +222,67 @@ def _sincedb_open
 
       @logger.debug("_sincedb_open: reading from #{path}")
       db.each do |line|
-        ino, dev_major, dev_minor, pos = line.split(" ", 4)
+        ino, dev_major, dev_minor, pos = line.split("*", 4)
         inode = [ino, dev_major.to_i, dev_minor.to_i]
         @logger.debug("_sincedb_open: setting #{inode.inspect} to #{pos.to_i}")
         @sincedb[inode] = pos.to_i
       end
+      db.close
     end # def _sincedb_open
 
     private
-    def _sincedb_write
+    def _sincedb_write_if_pending
+
+      #  Check to see if sincedb should be written out since there was a file read after the sincedb flush, 
+      #  and during the sincedb_write_interval
+
+      if @sincedb_write_pending
+        _sincedb_write
+      end
+    end
+
+    private
+    def _sincedb_write(sincedb_force_write=false)
+
+      # This routine will only write out sincedb if enough time has passed based on @sincedb_write_interval
+      # If it hasn't and we were asked to write, then we are pending a write.
+
+      # if we were called with force == true, then we have to write sincedb and bypass a time check 
+      # ie. external caller calling the public sincedb_write method
+
+      if(@sincedb_writing)
+        @logger.warn("_sincedb_write already writing")
+        return
+      end
+
+      @sincedb_writing = true
+
+      if (!sincedb_force_write)
+         now = Time.now.to_i
+         delta = now - @sincedb_last_write
+
+         # we will have to flush out the sincedb file after the interval expires.  So, we will try again later.
+         if delta < @opts[:sincedb_write_interval]
+           @sincedb_write_pending = true
+           @sincedb_writing = false
+           return
+         end
+      end
+
+      @logger.debug("writing sincedb (delta since last write = #{delta})")
+
       path = @opts[:sincedb_path]
       tmp = "#{path}.new"
       begin
         db = File.open(tmp, "w")
       rescue => e
         @logger.warn("_sincedb_write failed: #{tmp}: #{e}")
+        @sincedb_writing = false
         return
       end
 
       @sincedb.each do |inode, pos|
-        db.puts([inode, pos].flatten.join(" "))
+        db.puts([inode, pos].flatten.join("*"))
       end
       db.close
 
@@ -242,6 +291,12 @@ def _sincedb_write
       rescue => e
         @logger.warn("_sincedb_write rename/sync failed: #{tmp} -> #{path}: #{e}")
       end
+
+      @sincedb_last_write = now
+      @sincedb_write_pending = false
+      @sincedb_writing = false
+
+      System.gc()
     end # def _sincedb_write
 
     public

lib/filewatch/watch.rb

@@ -78,7 +78,7 @@ def each(&block)
         else
           inode = [stat.ino.to_s, stat.dev_major, stat.dev_minor]
         end
-		
+
         if inode != @files[path][:inode]
           @logger.debug("#{path}: old inode was #{@files[path][:inode].inspect}, new is #{inode.inspect}")
           yield(:delete, path)
@@ -90,6 +90,11 @@ def each(&block)
         elsif stat.size > @files[path][:size]
           @logger.debug("#{path}: file grew, old size #{@files[path][:size]}, new size #{stat.size}")
           yield(:modify, path)
+        else 
+          # since there is no update, we should pass control back in case the caller needs to do any work
+          # otherwise, they can ONLY do other work when a file is created or modified
+          @logger.debug("#{path}: nothing to update")
+          yield(:noupdate, path)
         end
 
         @files[path][:size] = stat.size
@@ -152,14 +157,14 @@ def _discover_file(path, initial=false)
           :inode => [stat.ino, stat.dev_major, stat.dev_minor],
           :create_sent => false,
         }
-		
-		if @iswindows
+
+        if @iswindows
           fileId = Winhelper.GetWindowsUniqueFileIdentifier(path)
           @files[file][:inode] = [fileId, stat.dev_major, stat.dev_minor]
         else
           @files[file][:inode] = [stat.ino.to_s, stat.dev_major, stat.dev_minor]
         end
-		
+
         if initial
           @files[file][:initial] = true
         end