1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
package clusterprovider
import (
"context"
"errors"
"fmt"
"path/filepath"
"github.com/fsnotify/fsnotify"
"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/fswatch"
"github.com/johnandersen777/socialweb-computer-kcp/internal/procgroup"
)
func timedOut(err error) bool { return errors.Is(err, ErrAwaitTimeout) }
type logged interface {
LogPath() string
}
func (m unitMember) LogPath() string { return m.path }
func untilOutput(ctx context.Context, m procgroup.Member, done func(out string) (bool, error), timeout func() error) error {
path := ""
if l, ok := m.(logged); ok {
path = filepath.Clean(l.LogPath())
}
var events <-chan fsnotify.Event
var errs <-chan error
if path != "" && path != "." {
w, err := fswatch.Shared().Subscribe(filepath.Dir(path))
if err != nil {
return fmt.Errorf("clusterprovider: watching the output of %s: %w", path, err)
}
defer w.Close()
events, errs = w.Events(), w.Errors()
}
for {
ok, err := done(m.Output())
if err != nil {
return err
}
if ok {
return nil
}
for wait := true; wait; {
select {
case <-ctx.Done():
return timeout()
case ev, open := <-events:
if !open {
events = nil
continue
}
wait = filepath.Clean(ev.Name) != path || ev.Op == fsnotify.Chmod
case _, open := <-errs:
if !open {
errs = nil
}
}
}
}
}
