(maint) allow event filtering and file filtering for file watching

This adds the ability for a callback to add filtering on specific
events, and specific file paths if desired, which will reduce the
number of events passed to the callbacks.

src/clj/puppetlabs/trapperkeeper/services/protocols/filesystem_watch_service.clj

@@ -25,7 +25,22 @@
     When dir is deleted, the behavior is unspecified, left up to the
     implementation, and may be platform-specific.")
 
-  (add-callback! [this callback]
+  (add-file-callback! [this path-to-file callback] [this path-to-file callback types]
+    "Adds a callback to a Watcher tha will be invoked when the watched file changes.
+    The parent directory containing the file must be added via `add-watch-dir!` for the callbacks to be triggered.
+    The callback will be passed a sequence of Events as its only argument.
+    The exact events passed to the callback are unspecified, left up to the implementation,
+    and possibly platform-dependent; however, the following events are guaranteed to be passed to the callback
+
+     * an event of :type :create with :path p, when a file is created at path p
+     * an event of :type :modify with :path p, when the contents of a file at path p are modified
+     * an event of :type :delete with :path p, when a file is deleted at path p
+
+     The types of events where the callback is triggered can be limited with the optional fourth argument. The fourth
+     argument should be the `set` of types of interest from: `create`, `modify`, `delete`, `unknown`.
+     The default is to specify all types.")
+
+  (add-callback! [this callback] [this callback types]
     "Adds a callback to a Watcher.  The callback will be invoked when any
     watched directories change.  The callback will be passed a sequence of
     Events as its only argument.  The exact events passed to the callback are
@@ -38,7 +53,11 @@
 
     Note that, for any of those particular changes, there may also be additional
     events passed to the callback, such as events on a parent directory of a
-    changed file."))
+    changed file.
+
+    The types of events where the callback is triggered can be limited with the optional third argument.
+    The types argument should be the `set` of types of interest from: `create`, `modify`, `delete`, `unknown`.
+    The default is to specify all types."))
 
 (defprotocol FilesystemWatchService
   (create-watcher [this] [this options]

src/clj/puppetlabs/trapperkeeper/services/watcher/filesystem_watch_core.clj

@@ -62,6 +62,8 @@
       (IllegalArgumentException.
         (trs "Must pass a boolean value for :recursive (directory watching) option")))))
 
+(def all-types #{:create :modify :delete :unknown})
+
 (defrecord WatcherImpl
   [watch-service callbacks recursive]
   Watcher
@@ -86,9 +88,21 @@
                @recursive (:recursive options)))))
     (watch-protocol/add-watch-dir! this dir))
 
+  (add-file-callback!
+    [this path-to-file callback]
+    (watch-protocol/add-file-callback! this path-to-file callback all-types))
+
+  (add-file-callback!
+    [_this path-to-file callback types]
+    (swap! callbacks conj {:callback callback :file path-to-file :types (set types)}))
+
+  (add-callback!
+    [this callback]
+    (watch-protocol/add-callback! this callback all-types))
+
   (add-callback!
-    [_this callback]
-    (swap! callbacks conj callback)))
+    [_this callback types]
+    (swap! callbacks conj {:callback callback :types (set types)})))
 
 (defn create-watcher
   ([]
@@ -120,7 +134,7 @@
     (map #(clojurize % (.watchable watch-key)) events)))
 
 (schema/defn retrieve-events :- [Event]
-  "Blocks until an event the watcher is concerned with has occured. Will then
+  "Blocks until an event the watcher is concerned with has occurred. Will then
   poll for a new event, waiting at least `window-min` for a new event to
   occur. Will continue polling for as long as there are new events that occur
   within `window-min`, or the `window-max` time limit has been exceeded."
@@ -146,6 +160,19 @@
           events))
       initial-events)))
 
+(defn filter-events
+  [callback-entry events]
+  (let [allowed-types (:types callback-entry)
+        event-filter (fn [event] (contains? allowed-types (:type event)))
+        ;; construct an optimal filter function based on the type of filtering needed
+        filter-fn (if (nil? (:file callback-entry))
+                    ;; check the type match only
+                    event-filter
+                    ;; check the file as well as the type match
+                    (fn [event] (and (event-filter event)
+                                     (= (.toPath (:changed-path event)) (:file callback-entry)))))]
+    (filter filter-fn events)))
+
 (schema/defn process-events!
   "Process for side effects any events that occurred for watcher's watch-key"
   [watcher :- (schema/protocol Watcher)
@@ -156,16 +183,14 @@
       (let [events-by-dir (group-by :watched-path events)]
         (doseq [[dir events'] events-by-dir]
           (log/debug (trs "Got {0} event(s) in directory {1}"
-                       (count events') dir)))))
-
+                          (count events') dir)))))
     ;; avoid doing a potentially expensive print-to-string when we aren't logging at :trace
     (when (log/enabled? :trace)
       (log/tracef "%s\n%s"
                   (trs "Events:")
                   (ks/pprint-to-string events)))
-
-    (doseq [callback callbacks]
-      (callback events))))
+    (doseq [callback-entry callbacks]
+      ((:callback callback-entry) (filter-events callback-entry events)))))
 
 (schema/defn watch! :- Future
   "Creates and returns a future. Processes events for the passed in watcher within the context of that future.