debounce refresh requests with quietperiod

Signed-off-by: Nicolas De Loof <[email protected]>

go.mod

@@ -81,6 +81,7 @@ require (
 	github.com/imdario/mergo v0.3.13 // indirect
 	github.com/inconshreveable/mousetrap v1.0.1 // indirect
 	github.com/jinzhu/gorm v1.9.11 // indirect
+	github.com/jonboulle/clockwork v0.3.1-0.20230117163003-a89700cec744
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51 // indirect
 	github.com/klauspost/compress v1.15.9 // indirect

go.sum

@@ -392,6 +392,8 @@ github.com/jinzhu/now v1.1.1/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/
 github.com/jmhodges/clock v0.0.0-20160418191101-880ee4c33548/go.mod h1:hGT6jSUVzF6no3QaDSMLGLEHtHSBSefs+MgcDWnmhmo=
 github.com/jmoiron/sqlx v0.0.0-20180124204410-05cef0741ade/go.mod h1:IiEW3SEiiErVyFdH8NTuWjSifiEQKUoyK3LNqr2kCHU=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
+github.com/jonboulle/clockwork v0.3.1-0.20230117163003-a89700cec744 h1:fJ+REXDOpsMqA2spt3wAq3HGJJvWnNitGK2KVZTos+8=
+github.com/jonboulle/clockwork v0.3.1-0.20230117163003-a89700cec744/go.mod h1:xgRqUGwRcjKCO1vbZUEtSLrqKoPSsUpK7fnezOII0kc=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=

pkg/compose/build.go

@@ -44,14 +44,16 @@ import (
 
 func (s *composeService) Build(ctx context.Context, project *types.Project, options api.BuildOptions) error {
 	return progress.Run(ctx, func(ctx context.Context) error {
-		return s.build(ctx, project, options)
+		_, err := s.build(ctx, project, options)
+		return err
 	})
 }
 
-func (s *composeService) build(ctx context.Context, project *types.Project, options api.BuildOptions) error {
+func (s *composeService) build(ctx context.Context, project *types.Project, options api.BuildOptions) (map[string]string, error) {
 	args := flatten(options.Args.Resolve(envResolver(project.Environment)))
 
-	return InDependencyOrder(ctx, project, func(ctx context.Context, name string) error {
+	var imageIDs map[string]string
+	err := InDependencyOrder(ctx, project, func(ctx context.Context, name string) error {
 		if len(options.Services) > 0 && !utils.Contains(options.Services, name) {
 			return nil
 		}
@@ -93,11 +95,12 @@ func (s *composeService) build(ctx context.Context, project *types.Project, opti
 			}}
 		}
 		opts := map[string]build.Options{imageName: buildOptions}
-		_, err = s.doBuild(ctx, project, opts, options.Progress)
+		imageIDs, err = s.doBuild(ctx, project, opts, options.Progress)
 		return err
 	}, func(traversal *graphTraversal) {
 		traversal.maxConcurrency = s.maxConcurrency
 	})
+	return imageIDs, err
 }
 
 func (s *composeService) ensureImagesExists(ctx context.Context, project *types.Project, quietPull bool) error {

pkg/compose/watch.go

@@ -18,10 +18,14 @@ import (
 	"context"
 	"fmt"
 	"log"
+	"strings"
+	"time"
 
 	"github.com/compose-spec/compose-go/types"
 	"github.com/docker/compose/v2/pkg/api"
+	"github.com/docker/compose/v2/pkg/utils"
 	"github.com/fsnotify/fsnotify"
+	"github.com/jonboulle/clockwork"
 	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -30,10 +34,47 @@ import (
 type DevelopmentConfig struct {
 }
 
+const quietPeriod = 2 * time.Second
+
 func (s *composeService) Watch(ctx context.Context, project *types.Project, services []string, options api.WatchOptions) error {
 	fmt.Fprintln(s.stderr(), "not implemented yet")
 
 	eg, ctx := errgroup.WithContext(ctx)
+	needRefresh := make(chan string)
+	eg.Go(func() error {
+		clock := clockwork.NewRealClock()
+		debounce(ctx, clock, quietPeriod, needRefresh, func(services []string) {
+			fmt.Fprintf(s.stderr(), "Updating %s after changes were detected\n", strings.Join(services, ", "))
+			imageIds, err := s.build(ctx, project, api.BuildOptions{
+				Services: services,
+			})
+			if err != nil {
+				fmt.Fprintf(s.stderr(), "Build failed")
+			}
+			for i, service := range project.Services {
+				if id, ok := imageIds[service.Name]; ok {
+					service.Image = id
+				}
+				project.Services[i] = service
+			}
+
+			err = s.Up(ctx, project, api.UpOptions{
+				Create: api.CreateOptions{
+					Services: services,
+					Inherit:  true,
+				},
+				Start: api.StartOptions{
+					Services: services,
+					Project:  project,
+				},
+			})
+			if err != nil {
+				fmt.Fprintf(s.stderr(), "Application failed to start after update")
+			}
+		})
+		return nil
+	})
+
 	err := project.WithServices(services, func(service types.ServiceConfig) error {
 		var config DevelopmentConfig
 		if y, ok := service.Extensions["x-develop"]; ok {
@@ -64,6 +105,7 @@ func (s *composeService) Watch(ctx context.Context, project *types.Project, serv
 					return nil
 				case event := <-watcher.Events:
 					log.Println("fs event :", event.String())
+					needRefresh <- service.Name
 				case err := <-watcher.Errors:
 					return err
 				}
@@ -77,3 +119,23 @@ func (s *composeService) Watch(ctx context.Context, project *types.Project, serv
 
 	return eg.Wait()
 }
+
+func debounce(ctx context.Context, clock clockwork.Clock, delay time.Duration, input chan string, fn func(services []string)) {
+	services := utils.Set[string]{}
+	t := clock.AfterFunc(delay, func() {
+		if len(services) > 0 {
+			refresh := services.Elements()
+			services.Clear()
+			fn(refresh)
+		}
+	})
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case service := <-input:
+			t.Reset(delay)
+			services.Add(service)
+		}
+	}
+}

pkg/compose/watch_test.go

@@ -0,0 +1,52 @@
+/*
+
+   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 compose
+
+import (
+	"context"
+	"testing"
+
+	"github.com/jonboulle/clockwork"
+	"golang.org/x/sync/errgroup"
+	"gotest.tools/v3/assert"
+)
+
+func Test_debounce(t *testing.T) {
+	ch := make(chan string)
+	var (
+		ran int
+		got []string
+	)
+	clock := clockwork.NewFakeClock()
+	ctx, stop := context.WithCancel(context.TODO())
+	eg, ctx := errgroup.WithContext(ctx)
+	eg.Go(func() error {
+		debounce(ctx, clock, quietPeriod, ch, func(services []string) {
+			got = append(got, services...)
+			ran++
+			stop()
+		})
+		return nil
+	})
+	for i := 0; i < 100; i++ {
+		ch <- "test"
+	}
+	assert.Equal(t, ran, 0)
+	clock.Advance(quietPeriod)
+	err := eg.Wait()
+	assert.NilError(t, err)
+	assert.Equal(t, ran, 1)
+	assert.DeepEqual(t, got, []string{"test"})
+}

pkg/utils/set.go

@@ -0,0 +1,39 @@
+/*
+
+   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 utils
+
+type Set[T comparable] map[T]struct{}
+
+func (s Set[T]) Add(v T) {
+	s[v] = struct{}{}
+}
+
+func (s Set[T]) Remove(v T) {
+	delete(s, v)
+}
+
+func (s Set[T]) Clear() {
+	for v := range s {
+		delete(s, v)
+	}
+}
+
+func (s Set[T]) Elements() []T {
+	elements := make([]T, 0, len(s))
+	for v := range s {
+		elements = append(elements, v)
+	}
+	return elements
+}