From fb088e62ac9d53881828e8530a20248de9c03425 Mon Sep 17 00:00:00 2001 From: David Gageot Date: Fri, 21 Dec 2018 08:18:07 +0100 Subject: [PATCH] Move trigger related code to the watcher Signed-off-by: David Gageot --- pkg/skaffold/runner/dev.go | 8 ++------ pkg/skaffold/runner/dev_test.go | 11 ++++++----- pkg/skaffold/runner/runner.go | 4 +--- pkg/skaffold/watch/watch.go | 28 ++++++++++++++++++---------- pkg/skaffold/watch/watch_test.go | 11 +++++------ 5 files changed, 32 insertions(+), 30 deletions(-) diff --git a/pkg/skaffold/runner/dev.go b/pkg/skaffold/runner/dev.go index 373cd6e1008..a80f7365aed 100644 --- a/pkg/skaffold/runner/dev.go +++ b/pkg/skaffold/runner/dev.go @@ -41,10 +41,7 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la // Create watcher and register artifacts to build current state of files. changed := changes{} onChange := func() error { - defer func() { - changed.reset() - r.Trigger.WatchForChanges(out) - }() + defer changed.reset() logger.Mute() @@ -149,6 +146,5 @@ func (r *SkaffoldRunner) Dev(ctx context.Context, out io.Writer, artifacts []*la } } - r.Trigger.WatchForChanges(out) - return r.Watcher.Run(ctx, r.Trigger, onChange) + return r.Watcher.Run(ctx, out, onChange) } diff --git a/pkg/skaffold/runner/dev_test.go b/pkg/skaffold/runner/dev_test.go index ae428a65835..13ea808ab4d 100644 --- a/pkg/skaffold/runner/dev_test.go +++ b/pkg/skaffold/runner/dev_test.go @@ -19,6 +19,7 @@ package runner import ( "context" "errors" + "io" "io/ioutil" "testing" @@ -29,21 +30,21 @@ import ( type NoopWatcher struct{} -func (t *NoopWatcher) Register(deps func() ([]string, error), onChange func(watch.Events)) error { +func (t *NoopWatcher) Register(func() ([]string, error), func(watch.Events)) error { return nil } -func (t *NoopWatcher) Run(ctx context.Context, trigger watch.Trigger, onChange func() error) error { +func (t *NoopWatcher) Run(context.Context, io.Writer, func() error) error { return nil } type FailWatcher struct{} -func (t *FailWatcher) Register(deps func() ([]string, error), onChange func(watch.Events)) error { +func (t *FailWatcher) Register(func() ([]string, error), func(watch.Events)) error { return nil } -func (t *FailWatcher) Run(ctx context.Context, trigger watch.Trigger, onChange func() error) error { +func (t *FailWatcher) Run(context.Context, io.Writer, func() error) error { return errors.New("BUG") } @@ -58,7 +59,7 @@ func (t *TestWatcher) Register(deps func() ([]string, error), onChange func(watc return nil } -func (t *TestWatcher) Run(ctx context.Context, trigger watch.Trigger, onChange func() error) error { +func (t *TestWatcher) Run(ctx context.Context, out io.Writer, onChange func() error) error { for _, evt := range t.events { t.testBench.enterNewCycle() diff --git a/pkg/skaffold/runner/runner.go b/pkg/skaffold/runner/runner.go index c411749a19a..22f35cfe68d 100644 --- a/pkg/skaffold/runner/runner.go +++ b/pkg/skaffold/runner/runner.go @@ -48,7 +48,6 @@ type SkaffoldRunner struct { deploy.Deployer test.Tester tag.Tagger - watch.Trigger sync.Syncer watch.Watcher @@ -107,9 +106,8 @@ func NewForConfig(opts *config.SkaffoldOptions, cfg *latest.SkaffoldPipeline) (* Tester: tester, Deployer: deployer, Tagger: tagger, - Trigger: trigger, Syncer: &kubectl.Syncer{}, - Watcher: watch.NewWatcher(), + Watcher: watch.NewWatcher(trigger), opts: opts, imageList: kubernetes.NewImageList(), }, nil diff --git a/pkg/skaffold/watch/watch.go b/pkg/skaffold/watch/watch.go index e61ec45842d..1a3b8a19205 100644 --- a/pkg/skaffold/watch/watch.go +++ b/pkg/skaffold/watch/watch.go @@ -18,6 +18,7 @@ package watch import ( "context" + "io" "github.com/pkg/errors" ) @@ -28,14 +29,19 @@ type Factory func() Watcher // Watcher monitors files changes for multiples components. type Watcher interface { Register(deps func() ([]string, error), onChange func(Events)) error - Run(ctx context.Context, trigger Trigger, onChange func() error) error + Run(ctx context.Context, out io.Writer, onChange func() error) error } -type watchList []*component +type watchList struct { + components []*component + trigger Trigger +} // NewWatcher creates a new Watcher. -func NewWatcher() Watcher { - return &watchList{} +func NewWatcher(trigger Trigger) Watcher { + return &watchList{ + trigger: trigger, + } } type component struct { @@ -52,7 +58,7 @@ func (w *watchList) Register(deps func() ([]string, error), onChange func(Events return errors.Wrap(err, "listing files") } - *w = append(*w, &component{ + w.components = append(w.components, &component{ deps: deps, onChange: onChange, state: state, @@ -61,19 +67,20 @@ func (w *watchList) Register(deps func() ([]string, error), onChange func(Events } // Run watches files until the context is cancelled or an error occurs. -func (w *watchList) Run(ctx context.Context, trigger Trigger, onChange func() error) error { - t, cleanup := trigger.Start() +func (w *watchList) Run(ctx context.Context, out io.Writer, onChange func() error) error { + t, cleanup := w.trigger.Start() defer cleanup() changedComponents := map[int]bool{} + w.trigger.WatchForChanges(out) for { select { case <-ctx.Done(): return nil case <-t: changed := 0 - for i, component := range *w { + for i, component := range w.components { state, err := Stat(component.deps) if err != nil { return errors.Wrap(err, "listing files") @@ -93,9 +100,9 @@ func (w *watchList) Run(ctx context.Context, trigger Trigger, onChange func() er // To prevent that, we debounce changes that happen too quickly // by waiting for a full turn where nothing happens and trigger a rebuild for // the accumulated changes. - debounce := trigger.Debounce() + debounce := w.trigger.Debounce() if (!debounce && changed > 0) || (debounce && changed == 0 && len(changedComponents) > 0) { - for i, component := range *w { + for i, component := range w.components { if changedComponents[i] { component.onChange(component.events) } @@ -106,6 +113,7 @@ func (w *watchList) Run(ctx context.Context, trigger Trigger, onChange func() er } changedComponents = map[int]bool{} + w.trigger.WatchForChanges(out) } } } diff --git a/pkg/skaffold/watch/watch_test.go b/pkg/skaffold/watch/watch_test.go index bf47a32f1ab..c1ba6838b2f 100644 --- a/pkg/skaffold/watch/watch_test.go +++ b/pkg/skaffold/watch/watch_test.go @@ -18,6 +18,7 @@ package watch import ( "context" + "io/ioutil" "sync" "testing" "time" @@ -70,7 +71,9 @@ func TestWatch(t *testing.T) { somethingChanged := newCallback() // Watch folder - watcher := NewWatcher() + watcher := NewWatcher(&pollTrigger{ + Interval: 10 * time.Millisecond, + }) err := watcher.Register(folder.List, folderChanged.call) testutil.CheckError(t, false, err) @@ -79,11 +82,7 @@ func TestWatch(t *testing.T) { var stopped sync.WaitGroup stopped.Add(1) go func() { - trigger := &pollTrigger{ - Interval: 10 * time.Millisecond, - } - - err = watcher.Run(ctx, trigger, somethingChanged.callNoErr) + err = watcher.Run(ctx, ioutil.Discard, somethingChanged.callNoErr) stopped.Done() testutil.CheckError(t, false, err) }()