feat(watcher): migrate to fsnotify (#2048)

internal/fingerprint/glob.go

@@ -12,28 +12,20 @@ import (
 )
 
 func Globs(dir string, globs []*ast.Glob) ([]string, error) {
-	fileMap := make(map[string]bool)
+	resultMap := make(map[string]bool)
 	for _, g := range globs {
-		matches, err := Glob(dir, g.Glob)
+		matches, err := glob(dir, g.Glob)
 		if err != nil {
 			continue
 		}
 		for _, match := range matches {
-			fileMap[match] = !g.Negate
+			resultMap[match] = !g.Negate
 		}
 	}
-	files := make([]string, 0)
-	for file, includePath := range fileMap {
-		if includePath {
-			files = append(files, file)
-		}
-	}
-	sort.Strings(files)
-	return files, nil
+	return collectKeys(resultMap), nil
 }
 
-func Glob(dir string, g string) ([]string, error) {
-	files := make([]string, 0)
+func glob(dir string, g string) ([]string, error) {
 	g = filepathext.SmartJoin(dir, g)
 
 	g, err := execext.Expand(g)
@@ -46,6 +38,8 @@ func Glob(dir string, g string) ([]string, error) {
 		return nil, err
 	}
 
+	results := make(map[string]bool, len(fs))
+
 	for _, f := range fs {
 		info, err := os.Stat(f)
 		if err != nil {
@@ -54,7 +48,18 @@ func Glob(dir string, g string) ([]string, error) {
 		if info.IsDir() {
 			continue
 		}
-		files = append(files, f)
+		results[f] = true
 	}
-	return files, nil
+	return collectKeys(results), nil
+}
+
+func collectKeys(m map[string]bool) []string {
+	keys := make([]string, 0, len(m))
+	for k, v := range m {
+		if v {
+			keys = append(keys, k)
+		}
+	}
+	sort.Strings(keys)
+	return keys
 }

internal/fingerprint/sources_checksum.go

@@ -56,7 +56,7 @@ func (checker *ChecksumChecker) IsUpToDate(t *ast.Task) (bool, error) {
 			if g.Negate {
 				continue
 			}
-			generates, err := Glob(t.Dir, g.Glob)
+			generates, err := glob(t.Dir, g.Glob)
 			if os.IsNotExist(err) {
 				return false, nil
 			}

internal/fsnotifyext/fsnotify_dedup.go

@@ -0,0 +1,56 @@
+package fsnotifyext
+
+import (
+	"math"
+	"sync"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+type Deduper struct {
+	w        *fsnotify.Watcher
+	waitTime time.Duration
+	mutex    sync.Mutex
+}
+
+func NewDeduper(w *fsnotify.Watcher, waitTime time.Duration) *Deduper {
+	return &Deduper{
+		w:        w,
+		waitTime: waitTime,
+	}
+}
+
+func (d *Deduper) GetChan() chan fsnotify.Event {
+	channel := make(chan fsnotify.Event)
+	timers := make(map[string]*time.Timer)
+
+	go func() {
+		for {
+			event, ok := <-d.w.Events
+			switch {
+			case !ok:
+				return
+			case event.Op == fsnotify.Chmod:
+				continue
+			}
+
+			d.mutex.Lock()
+			timer, ok := timers[event.String()]
+			d.mutex.Unlock()
+
+			if !ok {
+				timer = time.AfterFunc(math.MaxInt64, func() { channel <- event })
+				timer.Stop()
+
+				d.mutex.Lock()
+				timers[event.String()] = timer
+				d.mutex.Unlock()
+			}
+
+			timer.Reset(d.waitTime)
+		}
+	}()
+
+	return channel
+}