Compare commits

..

1 commit

Author SHA1 Message Date
4c338efd28 Add renovate.json
Some checks failed
go/folderwatcher/pipeline/pr-master There was a failure building this commit
2024-09-26 11:16:35 +00:00
9 changed files with 190 additions and 366 deletions

2
Jenkinsfile vendored
View file

@ -8,7 +8,7 @@ pipeline {
agent {
docker {
image 'docker.sportwanninger.de/infrastructure/golang-mingw:1.23.3'
image 'docker.sportwanninger.de/infrastructure/golang-mingw:1.23.1'
args "${dindArgs()}"
}
}

View file

@ -1,12 +1,10 @@
package folderwatcher
import (
"fmt"
"github.com/sirupsen/logrus"
"os"
"path/filepath"
"regexp"
"slices"
"sync"
"time"
@ -24,10 +22,8 @@ type FolderWatcher struct {
quit chan struct{}
mutex sync.Mutex
wg sync.WaitGroup
debounceWg sync.WaitGroup
debounceMap map[string]*time.Timer
debouncedChan chan string
existingList []string
enqueuedFiles int
}
@ -53,18 +49,7 @@ func NewFolderWatcher(conf Config, includeExisting bool, quit chan struct{}) (*F
}
}
if len(matchedFolders) == 0 {
return nil, fmt.Errorf("no folders found matching '%s'", conf.Folder)
}
for _, path := range matchedFolders {
stat, err := os.Stat(path)
if err != nil {
return nil, err
}
if !stat.IsDir() {
return nil, fmt.Errorf("%s is not a folder", path)
}
err = watcher.Add(path)
if err != nil {
return nil, err
@ -86,7 +71,6 @@ func NewFolderWatcher(conf Config, includeExisting bool, quit chan struct{}) (*F
quit: quit,
debounceMap: make(map[string]*time.Timer),
debouncedChan: make(chan string),
existingList: []string{},
}
if includeExisting {
@ -97,18 +81,10 @@ func NewFolderWatcher(conf Config, includeExisting bool, quit chan struct{}) (*F
}
for _, entry := range entries {
if !entry.IsDir() {
fullFilePath := filepath.Join(path, entry.Name())
if folderWatcher.matchesPattern(fullFilePath) {
folderWatcher.existingList = append(folderWatcher.existingList, fullFilePath)
}
folderWatcher.foundFileEvent(filepath.Join(path, entry.Name()))
}
}
}
slices.SortFunc(folderWatcher.existingList, func(a, b string) int {
infoA, _ := os.Stat(a)
infoB, _ := os.Stat(b)
return infoA.ModTime().Compare(infoB.ModTime())
})
}
return folderWatcher, nil
@ -119,14 +95,21 @@ type HandlerCallback func(path string, err error)
func (w *FolderWatcher) Watch(handler FileHandler, callback HandlerCallback) {
defer w.watcher.Close()
initialWait := sync.WaitGroup{}
handlerWg := sync.WaitGroup{}
handlerWg.Add(1)
// Start File Handling
go func() {
defer handlerWg.Done()
for file := range w.debouncedChan {
for stopped := false; !stopped; {
select {
case <-w.quit:
logger.Info("Stopping watcher")
stopped = true
case event, ok := <-w.watcher.Events:
if !ok {
stopped = true
break
}
logger.Debug("event:", event)
if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) {
w.foundFileEvent(event.Name)
}
case file, _ := <-w.debouncedChan:
triggerCallback, err := handler(file)
if err != nil {
logger.Warn("error during handle:", err)
@ -134,42 +117,7 @@ func (w *FolderWatcher) Watch(handler FileHandler, callback HandlerCallback) {
if triggerCallback {
callback(file, err)
}
w.wg.Done() // mark enqueued file as done
}
}()
if len(w.existingList) > 0 {
logger.Infof("got %d existing files", len(w.existingList))
initialWait.Add(1)
go func() {
defer initialWait.Done()
for _, path := range w.existingList {
w.enqueue(path)
}
}()
}
// Start Event Handling
for stopped := false; !stopped; {
select {
case <-w.quit:
logger.Info("Stopping watcher")
w.debounceWg.Wait() // wait for debounce to settle
w.wg.Wait() // wait for existing files to be handled
close(w.debouncedChan) //close the debounced chan to mark it as done
stopped = true
case event, ok := <-w.watcher.Events:
initialWait.Wait() // wait that all initial are enqueued, before handling of first event
if !ok {
stopped = true
break
}
logger.Debug("event:", event)
if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) {
if w.matchesPattern(event.Name) {
w.enqueueDebounce(event.Name)
}
}
w.wg.Done()
case err, ok := <-w.watcher.Errors:
if !ok {
stopped = true
@ -178,39 +126,31 @@ func (w *FolderWatcher) Watch(handler FileHandler, callback HandlerCallback) {
logger.Warn("error:", err)
}
}
handlerWg.Wait() // wait for filehandler to be closed, implies debouncedChan is closed
w.wg.Wait()
logger.Infof("watcher handled %d files", w.enqueuedFiles)
}
func (w *FolderWatcher) matchesPattern(fileName string) bool {
func (w *FolderWatcher) foundFileEvent(fileName string) {
baseName := filepath.Base(fileName)
return w.regex == nil || w.regex.MatchString(baseName)
}
func (w *FolderWatcher) enqueue(f string) {
w.wg.Add(1)
w.mutex.Lock()
defer w.mutex.Unlock()
if timer, ok := w.debounceMap[f]; ok && timer != nil {
timer.Stop()
if w.regex == nil || w.regex.MatchString(baseName) {
w.enqueueDebounce(fileName)
}
delete(w.debounceMap, f)
w.enqueuedFiles = w.enqueuedFiles + 1
w.debouncedChan <- f
}
func (w *FolderWatcher) enqueueDebounce(f string) {
logger.Debugf("Debounced Enqueue of %s", f)
w.mutex.Lock()
defer w.mutex.Unlock()
if timer, ok := w.debounceMap[f]; ok && timer != nil {
timer.Stop()
} else {
w.debounceWg.Add(1)
w.wg.Add(1)
}
w.debounceMap[f] = time.AfterFunc(DebounceTime, func() {
defer w.debounceWg.Done()
w.enqueue(f)
w.mutex.Lock()
defer w.mutex.Unlock()
delete(w.debounceMap, f)
w.enqueuedFiles = w.enqueuedFiles + 1
w.debouncedChan <- f
})
}

View file

@ -1,162 +0,0 @@
package folderwatcher
import (
"context"
"os"
"path/filepath"
"strconv"
"sync"
"testing"
"time"
)
func TestGlobbingN1(t *testing.T) {
tmpPath, err := os.MkdirTemp("", ".folderwatcher_test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(tmpPath)
quitChan := make(chan struct{})
_ = os.MkdirAll(filepath.Join(tmpPath, "g0"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g1"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g2"), os.ModePerm)
filesIn := []string{"A", "B", "C"}
var filesOut []string
wg := sync.WaitGroup{}
conf := Config{
Folder: filepath.Join(tmpPath, "*"),
}
watcher, err := NewFolderWatcher(conf, true, quitChan)
if err != nil {
t.Fatal(err)
}
go watcher.Watch(func(filePath string) (bool, error) {
if !filepath.IsAbs(filePath) {
t.Errorf("file %s is not an absolute path", filePath)
}
rel, err := filepath.Rel(tmpPath, filePath)
if err != nil {
return true, err
}
filesOut = append(filesOut, rel)
wg.Done()
return true, nil
}, func(s string, err error) {
})
for i, f := range filesIn {
wg.Add(1)
_ = os.WriteFile(filepath.Join(tmpPath, "g"+strconv.Itoa(i), f), []byte{0}, os.ModePerm)
}
finishedChan := make(chan struct{})
go func() {
wg.Wait()
finishedChan <- struct{}{}
}()
ctx, ctxCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer ctxCancel()
select {
case <-finishedChan:
case <-ctx.Done():
t.Error(ctx.Err())
}
quitChan <- struct{}{}
if len(filesOut) != len(filesIn) {
t.Errorf("filesOut length %d != %d", len(filesOut), len(filesIn))
}
for i, f := range filesIn {
if !Contains(filesOut, filepath.Join("g"+strconv.Itoa(i), f)) {
t.Errorf("File %s not found in %s", f, filesOut)
}
}
}
func TestGlobbingN2(t *testing.T) {
tmpPath, err := os.MkdirTemp("", ".folderwatcher_test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(tmpPath)
quitChan := make(chan struct{})
_ = os.MkdirAll(filepath.Join(tmpPath, "g0", "s0"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g0", "s2"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g1", "s1"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g1", "s3"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "gX"), os.ModePerm)
filesIn := []string{"A", "B", "C", "D"}
var filesOut []string
wg := sync.WaitGroup{}
conf := Config{
Folder: filepath.Join(tmpPath, "*", "*"),
}
watcher, err := NewFolderWatcher(conf, true, quitChan)
if err != nil {
t.Fatal(err)
}
go watcher.Watch(func(filePath string) (bool, error) {
if !filepath.IsAbs(filePath) {
t.Errorf("file %s is not an absolute path", filePath)
}
rel, err := filepath.Rel(tmpPath, filePath)
if err != nil {
return true, err
}
filesOut = append(filesOut, rel)
wg.Done()
return true, nil
}, func(s string, err error) {
})
for i, f := range filesIn {
wg.Add(1)
_ = os.WriteFile(filepath.Join(tmpPath, "g"+strconv.Itoa(i%2), "s"+strconv.Itoa(i), f), []byte{0}, os.ModePerm)
}
wg.Add(1)
_ = os.WriteFile(filepath.Join(tmpPath, "gX", "foo"), []byte{0}, os.ModePerm)
finishedChan := make(chan struct{})
go func() {
wg.Wait()
finishedChan <- struct{}{}
}()
ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer ctxCancel()
select {
case <-finishedChan:
case <-ctx.Done():
wg.Done() // ctx Will Timeout, because gX/foo will always be remaining
// t.Error(ctx.Err())
}
quitChan <- struct{}{}
if len(filesOut) != len(filesIn) {
t.Errorf("filesOut length %d != %d", len(filesOut), len(filesIn))
}
for i, f := range filesIn {
if !Contains(filesOut, filepath.Join("g"+strconv.Itoa(i%2), "s"+strconv.Itoa(i), f)) {
t.Errorf("File %s not found in %s", f, filesOut)
}
}
if Contains(filesOut, filepath.Join("gX", "foo")) {
t.Errorf("File %s found in %s", filepath.Join("gX", "foo"), filesOut)
}
}

View file

@ -1,22 +0,0 @@
package folderwatcher
import (
"os"
"testing"
)
func TestNotExistingFolder(t *testing.T) {
tmpPath, err := os.MkdirTemp("", ".folderwatcher_test")
if err != nil {
t.Fatal(err)
}
_ = os.RemoveAll(tmpPath)
conf := Config{
Folder: tmpPath,
}
_, err = NewFolderWatcher(conf, true, nil)
if err == nil {
t.Fatal("expected error")
}
}

View file

@ -1,87 +0,0 @@
package folderwatcher
import (
"context"
"fmt"
"os"
"path/filepath"
"sync"
"testing"
"time"
)
func TestOldFilesInCreationOrder(t *testing.T) {
tmpPath, err := os.MkdirTemp("", ".folderwatcher_test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(tmpPath)
quitChan := make(chan struct{})
filesIn := []string{"2", "C", "B", "A", "1"}
makeAbsFilePaths(t, tmpPath, &filesIn)
var filesOut []string
wg := sync.WaitGroup{}
conf := Config{
Folder: tmpPath,
}
for _, f := range filesIn {
wg.Add(1)
_ = os.WriteFile(f, []byte{0}, os.ModePerm)
time.Sleep(10 * time.Millisecond)
}
watcher, err := NewFolderWatcher(conf, true, quitChan)
if err != nil {
t.Fatal(err)
}
go watcher.Watch(func(filePath string) (bool, error) {
if !filepath.IsAbs(filePath) {
t.Errorf("file %s is not an absolute path", filePath)
}
filesOut = append(filesOut, filePath)
wg.Done()
return true, nil
}, func(s string, err error) {
})
finishedChan := make(chan struct{})
go func() {
wg.Wait()
finishedChan <- struct{}{}
}()
ctx, ctxCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer ctxCancel()
select {
case <-finishedChan:
case <-ctx.Done():
t.Error(ctx.Err())
}
quitChan <- struct{}{}
if len(filesOut) != len(filesIn) {
t.Errorf("filesOut length %d != %d", len(filesOut), len(filesIn))
}
filesInS := fmt.Sprintf("%s", filesIn)
filesOutS := fmt.Sprintf("%s", filesOut)
if filesInS != filesOutS {
t.Errorf("filesIn != filesOut %s != %s", filesIn, filesOut)
}
}
func makeAbsFilePaths(t *testing.T, rootPath string, filesIn *[]string) {
t.Helper()
for i, f := range *filesIn { //remap filesIn to Full Path
(*filesIn)[i] = filepath.Join(rootPath, f)
if !filepath.IsAbs((*filesIn)[i]) {
t.Fatalf("file %s is not an absolute path", (*filesIn)[i])
}
}
}

View file

@ -4,6 +4,7 @@ import (
"context"
"os"
"path/filepath"
"strconv"
"sync"
"testing"
"time"
@ -313,6 +314,157 @@ func TestOldAndNewFileHandlingWithoutExisting(t *testing.T) {
}
}
func TestGlobbingN1(t *testing.T) {
tmpPath, err := os.MkdirTemp("", ".folderwatcher_test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(tmpPath)
quitChan := make(chan struct{})
_ = os.MkdirAll(filepath.Join(tmpPath, "g0"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g1"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g2"), os.ModePerm)
filesIn := []string{"A", "B", "C"}
var filesOut []string
wg := sync.WaitGroup{}
conf := Config{
Folder: filepath.Join(tmpPath, "*"),
}
watcher, err := NewFolderWatcher(conf, true, quitChan)
if err != nil {
t.Fatal(err)
}
go watcher.Watch(func(filePath string) (bool, error) {
if !filepath.IsAbs(filePath) {
t.Errorf("file %s is not an absolute path", filePath)
}
rel, err := filepath.Rel(tmpPath, filePath)
if err != nil {
return true, err
}
filesOut = append(filesOut, rel)
wg.Done()
return true, nil
}, func(s string, err error) {
})
for i, f := range filesIn {
wg.Add(1)
_ = os.WriteFile(filepath.Join(tmpPath, "g"+strconv.Itoa(i), f), []byte{0}, os.ModePerm)
}
finishedChan := make(chan struct{})
go func() {
wg.Wait()
finishedChan <- struct{}{}
}()
ctx, ctxCancel := context.WithTimeout(context.Background(), 10*time.Second)
defer ctxCancel()
select {
case <-finishedChan:
case <-ctx.Done():
t.Error(ctx.Err())
}
quitChan <- struct{}{}
if len(filesOut) != len(filesIn) {
t.Errorf("filesOut length %d != %d", len(filesOut), len(filesIn))
}
for i, f := range filesIn {
if !Contains(filesOut, filepath.Join("g"+strconv.Itoa(i), f)) {
t.Errorf("File %s not found in %s", f, filesOut)
}
}
}
func TestGlobbingN2(t *testing.T) {
tmpPath, err := os.MkdirTemp("", ".folderwatcher_test")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(tmpPath)
quitChan := make(chan struct{})
_ = os.MkdirAll(filepath.Join(tmpPath, "g0", "s0"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g0", "s2"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g1", "s1"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "g1", "s3"), os.ModePerm)
_ = os.MkdirAll(filepath.Join(tmpPath, "gX"), os.ModePerm)
filesIn := []string{"A", "B", "C", "D"}
var filesOut []string
wg := sync.WaitGroup{}
conf := Config{
Folder: filepath.Join(tmpPath, "*", "*"),
}
watcher, err := NewFolderWatcher(conf, true, quitChan)
if err != nil {
t.Fatal(err)
}
go watcher.Watch(func(filePath string) (bool, error) {
if !filepath.IsAbs(filePath) {
t.Errorf("file %s is not an absolute path", filePath)
}
rel, err := filepath.Rel(tmpPath, filePath)
if err != nil {
return true, err
}
filesOut = append(filesOut, rel)
wg.Done()
return true, nil
}, func(s string, err error) {
})
for i, f := range filesIn {
wg.Add(1)
_ = os.WriteFile(filepath.Join(tmpPath, "g"+strconv.Itoa(i%2), "s"+strconv.Itoa(i), f), []byte{0}, os.ModePerm)
}
wg.Add(1)
_ = os.WriteFile(filepath.Join(tmpPath, "gX", "foo"), []byte{0}, os.ModePerm)
finishedChan := make(chan struct{})
go func() {
wg.Wait()
finishedChan <- struct{}{}
}()
ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
defer ctxCancel()
select {
case <-finishedChan:
case <-ctx.Done():
wg.Done() // ctx Will Timeout, because gX/foo will always be remaining
// t.Error(ctx.Err())
}
quitChan <- struct{}{}
if len(filesOut) != len(filesIn) {
t.Errorf("filesOut length %d != %d", len(filesOut), len(filesIn))
}
for i, f := range filesIn {
if !Contains(filesOut, filepath.Join("g"+strconv.Itoa(i%2), "s"+strconv.Itoa(i), f)) {
t.Errorf("File %s not found in %s", f, filesOut)
}
}
if Contains(filesOut, filepath.Join("gX", "foo")) {
t.Errorf("File %s found in %s", filepath.Join("gX", "foo"), filesOut)
}
}
func Contains[T comparable](slice []T, item T) bool {
for _, v := range slice {
if v == item {

6
go.mod
View file

@ -1,10 +1,10 @@
module git.sportwanninger.de/go/folderwatcher
go 1.23.3
go 1.23.1
require (
github.com/fsnotify/fsnotify v1.8.0
github.com/fsnotify/fsnotify v1.7.0
github.com/sirupsen/logrus v1.9.3
)
require golang.org/x/sys v0.27.0 // indirect
require golang.org/x/sys v0.4.0 // indirect

8
go.sum
View file

@ -1,8 +1,8 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/fsnotify/fsnotify v1.8.0 h1:dAwr6QBTBZIkG8roQaJjGof0pp0EeF+tNV7YBP3F/8M=
github.com/fsnotify/fsnotify v1.8.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
@ -11,8 +11,8 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.27.0 h1:wBqf8DvsY9Y/2P8gAfPDEYNuS30J4lPHJxXSb/nJZ+s=
golang.org/x/sys v0.27.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.4.0 h1:Zr2JFtRQNX3BCZ8YtxRE9hNJYC8J6I1MVbMg6owUp18=
golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

3
renovate.json Normal file
View file

@ -0,0 +1,3 @@
{
"$schema": "https://docs.renovatebot.com/renovate-schema.json"
}