compose/pkg/watch/watcher_fsevent.go

//go:build darwin
// +build darwin

/*
   Copyright 2020 Docker Compose CLI authors

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

package watch

import (
	"path/filepath"
	"time"

	"github.com/fsnotify/fsevents"
	"github.com/pkg/errors"
	"github.com/sirupsen/logrus"
)

// A file watcher optimized for Darwin.
// Uses FSEvents to avoid the terrible perf characteristics of kqueue. Requires CGO
type fseventNotify struct {
	stream *fsevents.EventStream
	events chan FileEvent
	errors chan error
	stop   chan struct{}

	pathsWereWatching map[string]interface{}
	ignore            PathMatcher
	sawAnyHistoryDone bool
}

func (d *fseventNotify) loop() {
	for {
		select {
		case <-d.stop:
			return
		case events, ok := <-d.stream.Events:
			if !ok {
				return
			}

			for _, e := range events {
				e.Path = filepath.Join("/", e.Path)

				if e.Flags&fsevents.HistoryDone == fsevents.HistoryDone {
					d.sawAnyHistoryDone = true
					continue
				}

				// We wait until we've seen the HistoryDone event for this watcher before processing any events
				// so that we skip all of the "spurious" events that precede it.
				if !d.sawAnyHistoryDone {
					continue
				}

				_, isPathWereWatching := d.pathsWereWatching[e.Path]
				if e.Flags&fsevents.ItemIsDir == fsevents.ItemIsDir && e.Flags&fsevents.ItemCreated == fsevents.ItemCreated && isPathWereWatching {
					// This is the first create for the path that we're watching. We always get exactly one of these
					// even after we get the HistoryDone event. Skip it.
					continue
				}

				ignore, err := d.ignore.Matches(e.Path)
				if err != nil {
					logrus.Infof("Error matching path %q: %v", e.Path, err)
				} else if ignore {
					continue
				}

				d.events <- NewFileEvent(e.Path)
			}
		}
	}
}

// Add a path to be watched. Should only be called during initialization.
func (d *fseventNotify) initAdd(name string) {
	d.stream.Paths = append(d.stream.Paths, name)

	if d.pathsWereWatching == nil {
		d.pathsWereWatching = make(map[string]interface{})
	}
	d.pathsWereWatching[name] = struct{}{}
}

func (d *fseventNotify) Start() error {
	if len(d.stream.Paths) == 0 {
		return nil
	}

	numberOfWatches.Add(int64(len(d.stream.Paths)))

	d.stream.Start()

	go d.loop()

	return nil
}

func (d *fseventNotify) Close() error {
	numberOfWatches.Add(int64(-len(d.stream.Paths)))

	d.stream.Stop()
	close(d.errors)
	close(d.stop)

	return nil
}

func (d *fseventNotify) Events() chan FileEvent {
	return d.events
}

func (d *fseventNotify) Errors() chan error {
	return d.errors
}

func newFSEventWatcher(paths []string, ignore PathMatcher) (*fseventNotify, error) {
	dw := &fseventNotify{
		ignore: ignore,
		stream: &fsevents.EventStream{
			Latency: 1 * time.Millisecond,
			Flags:   fsevents.FileEvents,
			// NOTE(dmiller): this corresponds to the `sinceWhen` parameter in FSEventStreamCreate
			// https://developer.apple.com/documentation/coreservices/1443980-fseventstreamcreate
			EventID: fsevents.LatestEventID(),
		},
		events: make(chan FileEvent),
		errors: make(chan error),
		stop:   make(chan struct{}),
	}

	paths = dedupePathsForRecursiveWatcher(paths)
	for _, path := range paths {
		path, err := filepath.Abs(path)
		if err != nil {
			return nil, errors.Wrap(err, "newWatcher")
		}
		dw.initAdd(path)
	}

	return dw, nil
}

var _ Notify = &fseventNotify{}