You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

157 lines
3.5 KiB

package startup
import (
"context"
"sync"
"time"
)
type Stage struct {
ID string `json:"id"`
Label string `json:"label"`
State string `json:"state"`
}
type Status struct {
ClockSyncToken string `json:"clock_sync_token,omitempty"`
Ready bool `json:"ready"`
State string `json:"state"`
ElapsedMS int64 `json:"elapsed_ms"`
Completed int `json:"completed"`
Total int `json:"total"`
Stages []Stage `json:"stages"`
}
type Tracker struct {
mu sync.Mutex
started time.Time
finished time.Time
stages []Stage
}
func New(started time.Time) *Tracker { return &Tracker{started: started} }
func (t *Tracker) Add(id, label string) {
t.mu.Lock()
defer t.mu.Unlock()
t.stages = append(t.stages, Stage{ID: id, Label: label, State: "pending"})
}
func (t *Tracker) Set(id, state string) {
t.mu.Lock()
defer t.mu.Unlock()
for i := range t.stages {
if t.stages[i].ID == id {
t.stages[i].State = state
}
}
done := len(t.stages) > 0
for _, s := range t.stages {
if s.State != "ready" {
done = false
}
}
if done && t.finished.IsZero() {
t.finished = time.Now()
}
}
// Run bounds the readiness decision even if a device call fails to return.
// A late result can never overwrite the recorded timeout.
func (t *Tracker) Run(id string, timeout time.Duration, work func() error) error {
t.Set(id, "running")
result := make(chan error, 1)
go func() { result <- work() }()
timer := time.NewTimer(timeout)
defer timer.Stop()
var err error
select {
case err = <-result:
case <-timer.C:
err = context.DeadlineExceeded
}
if err != nil {
t.Set(id, "failed")
} else {
t.Set(id, "ready")
}
return err
}
// Retry never overlaps attempts, including when a device call stops responding.
// Delayed means work is still pending, not that it succeeded or was abandoned.
func (t *Tracker) Retry(ctx context.Context, id string, delayedAfter, interval time.Duration, work func() error, onError func(error)) error {
t.Set(id, "running")
delay := time.NewTimer(delayedAfter)
defer delay.Stop()
delayed := false
markDelayed := func() { delayed = true; t.Set(id, "delayed") }
for {
if err := ctx.Err(); err != nil {
return err
}
result := make(chan error, 1)
go func() { result <- work() }()
var err error
wait:
for {
select {
case <-ctx.Done():
return ctx.Err()
case <-delay.C:
markDelayed()
case err = <-result:
break wait
}
}
if ctx.Err() != nil {
return ctx.Err()
}
if err == nil {
t.Set(id, "ready")
return nil
}
if onError != nil {
onError(err)
}
if !delayed {
t.Set(id, "retrying")
}
pause := time.NewTimer(interval)
select {
case <-ctx.Done():
pause.Stop()
return ctx.Err()
case <-delay.C:
markDelayed()
select {
case <-ctx.Done():
pause.Stop()
return ctx.Err()
case <-pause.C:
}
case <-pause.C:
}
}
}
func (t *Tracker) Snapshot() Status {
t.mu.Lock()
defer t.mu.Unlock()
end := time.Now()
if !t.finished.IsZero() {
end = t.finished
}
s := Status{State: "starting", ElapsedMS: end.Sub(t.started).Milliseconds(), Total: len(t.stages), Stages: append([]Stage(nil), t.stages...)}
for _, stage := range t.stages {
if stage.State == "ready" {
s.Completed++
}
if stage.State == "failed" {
s.State = "failed"
}
if stage.State == "delayed" && s.State != "failed" {
s.State = "delayed"
}
}
s.Ready = s.Total > 0 && s.Completed == s.Total
if s.Ready {
s.State = "ready"
}
return s
}