chore(updater): internal/updater/loop subpackage
- Do not export updater interface - Export updater struct - Define local interfaces where needed - More restrictive updater loop interface in http control server - Inject `Updater` into updater loop as an interface
This commit is contained in:
Quentin McGaw
212
internal/updater/loop/loop.go
Normal file
212
internal/updater/loop/loop.go
Normal file
@@ -0,0 +1,212 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/qdm12/gluetun/internal/configuration/settings"
|
||||
"github.com/qdm12/gluetun/internal/constants"
|
||||
"github.com/qdm12/gluetun/internal/models"
|
||||
"github.com/qdm12/gluetun/internal/storage"
|
||||
"github.com/qdm12/gluetun/internal/updater"
|
||||
)
|
||||
|
||||
type Looper interface {
|
||||
Run(ctx context.Context, done chan<- struct{})
|
||||
RunRestartTicker(ctx context.Context, done chan<- struct{})
|
||||
GetStatus() (status models.LoopStatus)
|
||||
SetStatus(ctx context.Context, status models.LoopStatus) (
|
||||
outcome string, err error)
|
||||
GetSettings() (settings settings.Updater)
|
||||
SetSettings(settings settings.Updater) (outcome string)
|
||||
}
|
||||
|
||||
type Updater interface {
|
||||
UpdateServers(ctx context.Context) (allServers models.AllServers, err error)
|
||||
}
|
||||
|
||||
type looper struct {
|
||||
state state
|
||||
// Objects
|
||||
updater Updater
|
||||
flusher storage.Flusher
|
||||
setAllServers func(allServers models.AllServers)
|
||||
logger Logger
|
||||
// Internal channels and locks
|
||||
loopLock sync.Mutex
|
||||
start chan struct{}
|
||||
running chan models.LoopStatus
|
||||
stop chan struct{}
|
||||
stopped chan struct{}
|
||||
updateTicker chan struct{}
|
||||
backoffTime time.Duration
|
||||
// Mock functions
|
||||
timeNow func() time.Time
|
||||
timeSince func(time.Time) time.Duration
|
||||
}
|
||||
|
||||
const defaultBackoffTime = 5 * time.Second
|
||||
|
||||
type Logger interface {
|
||||
Info(s string)
|
||||
Warn(s string)
|
||||
Error(s string)
|
||||
}
|
||||
|
||||
func NewLooper(settings settings.Updater, currentServers models.AllServers,
|
||||
flusher storage.Flusher, setAllServers func(allServers models.AllServers),
|
||||
client *http.Client, logger Logger) Looper {
|
||||
return &looper{
|
||||
state: state{
|
||||
status: constants.Stopped,
|
||||
settings: settings,
|
||||
},
|
||||
updater: updater.New(settings, client, currentServers, logger),
|
||||
flusher: flusher,
|
||||
setAllServers: setAllServers,
|
||||
logger: logger,
|
||||
start: make(chan struct{}),
|
||||
running: make(chan models.LoopStatus),
|
||||
stop: make(chan struct{}),
|
||||
stopped: make(chan struct{}),
|
||||
updateTicker: make(chan struct{}),
|
||||
timeNow: time.Now,
|
||||
timeSince: time.Since,
|
||||
backoffTime: defaultBackoffTime,
|
||||
}
|
||||
}
|
||||
|
||||
func (l *looper) logAndWait(ctx context.Context, err error) {
|
||||
if err != nil {
|
||||
l.logger.Error(err.Error())
|
||||
}
|
||||
l.logger.Info("retrying in " + l.backoffTime.String())
|
||||
timer := time.NewTimer(l.backoffTime)
|
||||
l.backoffTime *= 2
|
||||
select {
|
||||
case <-timer.C:
|
||||
case <-ctx.Done():
|
||||
if !timer.Stop() {
|
||||
<-timer.C
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *looper) Run(ctx context.Context, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
crashed := false
|
||||
select {
|
||||
case <-l.start:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
for ctx.Err() == nil {
|
||||
updateCtx, updateCancel := context.WithCancel(ctx)
|
||||
|
||||
serversCh := make(chan models.AllServers)
|
||||
errorCh := make(chan error)
|
||||
runWg := &sync.WaitGroup{}
|
||||
runWg.Add(1)
|
||||
go func() {
|
||||
defer runWg.Done()
|
||||
servers, err := l.updater.UpdateServers(updateCtx)
|
||||
if err != nil {
|
||||
if updateCtx.Err() == nil {
|
||||
errorCh <- err
|
||||
}
|
||||
return
|
||||
}
|
||||
serversCh <- servers
|
||||
}()
|
||||
|
||||
if !crashed {
|
||||
l.running <- constants.Running
|
||||
crashed = false
|
||||
} else {
|
||||
l.backoffTime = defaultBackoffTime
|
||||
l.state.setStatusWithLock(constants.Running)
|
||||
}
|
||||
|
||||
stayHere := true
|
||||
for stayHere {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
updateCancel()
|
||||
runWg.Wait()
|
||||
close(errorCh)
|
||||
return
|
||||
case <-l.start:
|
||||
l.logger.Info("starting")
|
||||
updateCancel()
|
||||
runWg.Wait()
|
||||
stayHere = false
|
||||
case <-l.stop:
|
||||
l.logger.Info("stopping")
|
||||
updateCancel()
|
||||
runWg.Wait()
|
||||
l.stopped <- struct{}{}
|
||||
case servers := <-serversCh:
|
||||
l.setAllServers(servers)
|
||||
if err := l.flusher.FlushToFile(&servers); err != nil {
|
||||
l.logger.Error(err.Error())
|
||||
}
|
||||
runWg.Wait()
|
||||
l.state.setStatusWithLock(constants.Completed)
|
||||
l.logger.Info("Updated servers information")
|
||||
case err := <-errorCh:
|
||||
close(serversCh)
|
||||
runWg.Wait()
|
||||
l.state.setStatusWithLock(constants.Crashed)
|
||||
l.logAndWait(ctx, err)
|
||||
crashed = true
|
||||
stayHere = false
|
||||
}
|
||||
}
|
||||
updateCancel()
|
||||
close(errorCh)
|
||||
}
|
||||
}
|
||||
|
||||
func (l *looper) RunRestartTicker(ctx context.Context, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
timer := time.NewTimer(time.Hour)
|
||||
timer.Stop()
|
||||
timerIsStopped := true
|
||||
if period := *l.GetSettings().Period; period > 0 {
|
||||
timerIsStopped = false
|
||||
timer.Reset(period)
|
||||
}
|
||||
lastTick := time.Unix(0, 0)
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if !timerIsStopped && !timer.Stop() {
|
||||
<-timer.C
|
||||
}
|
||||
return
|
||||
case <-timer.C:
|
||||
lastTick = l.timeNow()
|
||||
l.start <- struct{}{}
|
||||
timer.Reset(*l.GetSettings().Period)
|
||||
case <-l.updateTicker:
|
||||
if !timerIsStopped && !timer.Stop() {
|
||||
<-timer.C
|
||||
}
|
||||
timerIsStopped = true
|
||||
period := *l.GetSettings().Period
|
||||
if period == 0 {
|
||||
continue
|
||||
}
|
||||
var waited time.Duration
|
||||
if lastTick.UnixNano() > 0 {
|
||||
waited = l.timeSince(lastTick)
|
||||
}
|
||||
leftToWait := period - waited
|
||||
timer.Reset(leftToWait)
|
||||
timerIsStopped = false
|
||||
}
|
||||
}
|
||||
}
|
||||
103
internal/updater/loop/state.go
Normal file
103
internal/updater/loop/state.go
Normal file
@@ -0,0 +1,103 @@
|
||||
package loop
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sync"
|
||||
|
||||
"github.com/qdm12/gluetun/internal/configuration/settings"
|
||||
"github.com/qdm12/gluetun/internal/constants"
|
||||
"github.com/qdm12/gluetun/internal/models"
|
||||
)
|
||||
|
||||
type state struct {
|
||||
status models.LoopStatus
|
||||
settings settings.Updater
|
||||
statusMu sync.RWMutex
|
||||
periodMu sync.RWMutex
|
||||
}
|
||||
|
||||
func (s *state) setStatusWithLock(status models.LoopStatus) {
|
||||
s.statusMu.Lock()
|
||||
defer s.statusMu.Unlock()
|
||||
s.status = status
|
||||
}
|
||||
|
||||
func (l *looper) GetStatus() (status models.LoopStatus) {
|
||||
l.state.statusMu.RLock()
|
||||
defer l.state.statusMu.RUnlock()
|
||||
return l.state.status
|
||||
}
|
||||
|
||||
var ErrInvalidStatus = errors.New("invalid status")
|
||||
|
||||
func (l *looper) SetStatus(ctx context.Context, status models.LoopStatus) (outcome string, err error) {
|
||||
l.state.statusMu.Lock()
|
||||
defer l.state.statusMu.Unlock()
|
||||
existingStatus := l.state.status
|
||||
|
||||
switch status {
|
||||
case constants.Running:
|
||||
switch existingStatus {
|
||||
case constants.Starting, constants.Running, constants.Stopping, constants.Crashed:
|
||||
return fmt.Sprintf("already %s", existingStatus), nil
|
||||
}
|
||||
l.loopLock.Lock()
|
||||
defer l.loopLock.Unlock()
|
||||
l.state.status = constants.Starting
|
||||
l.state.statusMu.Unlock()
|
||||
l.start <- struct{}{}
|
||||
|
||||
newStatus := constants.Starting // for canceled context
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case newStatus = <-l.running:
|
||||
}
|
||||
l.state.statusMu.Lock()
|
||||
l.state.status = newStatus
|
||||
return newStatus.String(), nil
|
||||
case constants.Stopped:
|
||||
switch existingStatus {
|
||||
case constants.Stopped, constants.Stopping, constants.Starting, constants.Crashed:
|
||||
return fmt.Sprintf("already %s", existingStatus), nil
|
||||
}
|
||||
l.loopLock.Lock()
|
||||
defer l.loopLock.Unlock()
|
||||
l.state.status = constants.Stopping
|
||||
l.state.statusMu.Unlock()
|
||||
l.stop <- struct{}{}
|
||||
|
||||
newStatus := constants.Stopping // for canceled context
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-l.stopped:
|
||||
newStatus = constants.Stopped
|
||||
}
|
||||
l.state.statusMu.Lock()
|
||||
l.state.status = newStatus
|
||||
return status.String(), nil
|
||||
default:
|
||||
return "", fmt.Errorf("%w: %s: it can only be one of: %s, %s",
|
||||
ErrInvalidStatus, status, constants.Running, constants.Stopped)
|
||||
}
|
||||
}
|
||||
|
||||
func (l *looper) GetSettings() (settings settings.Updater) {
|
||||
l.state.periodMu.RLock()
|
||||
defer l.state.periodMu.RUnlock()
|
||||
return l.state.settings
|
||||
}
|
||||
|
||||
func (l *looper) SetSettings(settings settings.Updater) (outcome string) {
|
||||
l.state.periodMu.Lock()
|
||||
defer l.state.periodMu.Unlock()
|
||||
settingsUnchanged := reflect.DeepEqual(settings, l.state.settings)
|
||||
if settingsUnchanged {
|
||||
return "settings left unchanged"
|
||||
}
|
||||
l.state.settings = settings
|
||||
l.updateTicker <- struct{}{}
|
||||
return "settings updated"
|
||||
}
|
||||
Reference in New Issue
Block a user