diff --git a/go.mod b/go.mod index ba6b079..0484b5b 100644 --- a/go.mod +++ b/go.mod @@ -1,10 +1,11 @@ module cln -go 1.21.7 +go 1.23.6 require gopkg.in/yaml.v3 v3.0.1 require ( + git.site.quack-lab.dev/dave/cyutils v1.4.0 github.com/bmatcuk/doublestar/v4 v4.8.1 github.com/stretchr/testify v1.11.1 ) @@ -12,4 +13,5 @@ require ( require ( github.com/davecgh/go-spew v1.1.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + golang.org/x/time v0.12.0 // indirect ) diff --git a/go.sum b/go.sum index b69818a..9007e73 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,5 @@ +git.site.quack-lab.dev/dave/cyutils v1.4.0 h1:/Xo3QfLIFNab+axHneWmUK4MyfuObl+qq8whF9vTQpk= +git.site.quack-lab.dev/dave/cyutils v1.4.0/go.mod h1:fBjALu2Cp2u2bDr+E4zbGVMBeIgFzROg+4TCcTNAiQU= github.com/bmatcuk/doublestar/v4 v4.8.1 h1:54Bopc5c2cAvhLRAzqOGCYHYyhcDHsFF4wWIR5wKP38= github.com/bmatcuk/doublestar/v4 v4.8.1/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= @@ -6,6 +8,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= +golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/main.go b/main.go index ad9363d..b6b3891 100644 --- a/main.go +++ b/main.go @@ -7,8 +7,8 @@ import ( "log" "os" "path/filepath" - "sync" "sync/atomic" + utils "git.site.quack-lab.dev/dave/cyutils" ) const deliminer = "," @@ -130,28 +130,35 @@ func handleStatusErrors(status chan error) { } } -// processInstructions processes all instructions from the channel +// processInstructions processes all instructions from the channel using parallel workers func processInstructions(instructions chan *LinkInstruction) int32 { var instructionsDone int32 = 0 - var wg sync.WaitGroup + + // Collect all instructions first + var allInstructions []*LinkInstruction for { instruction, ok := <-instructions if !ok { LogInfo("No more instructions to process") break } + allInstructions = append(allInstructions, instruction) + } + + // Process instructions in parallel using cyutils.WithWorkers + // Let the library handle worker count - use 4 workers as a reasonable default + utils.WithWorkers(4, allInstructions, func(workerID int, _ int, instruction *LinkInstruction) { LogInfo("Processing: %s", instruction.String()) status := make(chan error) go instruction.RunAsync(status) - wg.Add(1) err := <-status if err != nil { LogError("Failed processing instruction: %v", err) + } else { + atomic.AddInt32(&instructionsDone, 1) } - atomic.AddInt32(&instructionsDone, 1) - wg.Done() - } - wg.Wait() + }) + return instructionsDone }