Maint: split httpproxy files
This commit is contained in:
@@ -3,7 +3,6 @@ package httpproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -52,70 +51,6 @@ func NewLooper(logger logging.Logger, settings configuration.HTTPProxy) Looper {
|
||||
}
|
||||
}
|
||||
|
||||
func (l *looper) Run(ctx context.Context, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
|
||||
crashed := false
|
||||
|
||||
if l.GetSettings().Enabled {
|
||||
go func() {
|
||||
_, _ = l.SetStatus(ctx, constants.Running)
|
||||
}()
|
||||
}
|
||||
|
||||
select {
|
||||
case <-l.start:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
for ctx.Err() == nil {
|
||||
runCtx, runCancel := context.WithCancel(ctx)
|
||||
|
||||
settings := l.GetSettings()
|
||||
address := fmt.Sprintf(":%d", settings.Port)
|
||||
server := New(runCtx, address, l.logger, settings.Stealth, settings.Log, settings.User, settings.Password)
|
||||
|
||||
errorCh := make(chan error)
|
||||
go server.Run(runCtx, errorCh)
|
||||
|
||||
// TODO stable timer, check Shadowsocks
|
||||
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():
|
||||
runCancel()
|
||||
<-errorCh
|
||||
return
|
||||
case <-l.start:
|
||||
l.logger.Info("starting")
|
||||
runCancel()
|
||||
<-errorCh
|
||||
stayHere = false
|
||||
case <-l.stop:
|
||||
l.logger.Info("stopping")
|
||||
runCancel()
|
||||
<-errorCh
|
||||
l.stopped <- struct{}{}
|
||||
case err := <-errorCh:
|
||||
l.state.setStatusWithLock(constants.Crashed)
|
||||
l.logAndWait(ctx, err)
|
||||
crashed = true
|
||||
stayHere = false
|
||||
}
|
||||
}
|
||||
runCancel() // repetition for linter only
|
||||
}
|
||||
}
|
||||
|
||||
func (l *looper) logAndWait(ctx context.Context, err error) {
|
||||
l.logger.Error(err.Error())
|
||||
l.logger.Info("retrying in " + l.backoffTime.String())
|
||||
|
||||
72
internal/httpproxy/run.go
Normal file
72
internal/httpproxy/run.go
Normal file
@@ -0,0 +1,72 @@
|
||||
package httpproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/qdm12/gluetun/internal/constants"
|
||||
)
|
||||
|
||||
func (l *looper) Run(ctx context.Context, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
|
||||
crashed := false
|
||||
|
||||
if l.GetSettings().Enabled {
|
||||
go func() {
|
||||
_, _ = l.SetStatus(ctx, constants.Running)
|
||||
}()
|
||||
}
|
||||
|
||||
select {
|
||||
case <-l.start:
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
|
||||
for ctx.Err() == nil {
|
||||
runCtx, runCancel := context.WithCancel(ctx)
|
||||
|
||||
settings := l.GetSettings()
|
||||
address := fmt.Sprintf(":%d", settings.Port)
|
||||
server := New(runCtx, address, l.logger, settings.Stealth, settings.Log, settings.User, settings.Password)
|
||||
|
||||
errorCh := make(chan error)
|
||||
go server.Run(runCtx, errorCh)
|
||||
|
||||
// TODO stable timer, check Shadowsocks
|
||||
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():
|
||||
runCancel()
|
||||
<-errorCh
|
||||
return
|
||||
case <-l.start:
|
||||
l.logger.Info("starting")
|
||||
runCancel()
|
||||
<-errorCh
|
||||
stayHere = false
|
||||
case <-l.stop:
|
||||
l.logger.Info("stopping")
|
||||
runCancel()
|
||||
<-errorCh
|
||||
l.stopped <- struct{}{}
|
||||
case err := <-errorCh:
|
||||
l.state.setStatusWithLock(constants.Crashed)
|
||||
l.logAndWait(ctx, err)
|
||||
crashed = true
|
||||
stayHere = false
|
||||
}
|
||||
}
|
||||
runCancel() // repetition for linter only
|
||||
}
|
||||
}
|
||||
41
internal/httpproxy/settings.go
Normal file
41
internal/httpproxy/settings.go
Normal file
@@ -0,0 +1,41 @@
|
||||
package httpproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
|
||||
"github.com/qdm12/gluetun/internal/configuration"
|
||||
"github.com/qdm12/gluetun/internal/constants"
|
||||
)
|
||||
|
||||
func (l *looper) GetSettings() (settings configuration.HTTPProxy) {
|
||||
l.state.settingsMu.RLock()
|
||||
defer l.state.settingsMu.RUnlock()
|
||||
return l.state.settings
|
||||
}
|
||||
|
||||
func (l *looper) SetSettings(ctx context.Context, settings configuration.HTTPProxy) (
|
||||
outcome string) {
|
||||
l.state.settingsMu.Lock()
|
||||
settingsUnchanged := reflect.DeepEqual(settings, l.state.settings)
|
||||
if settingsUnchanged {
|
||||
l.state.settingsMu.Unlock()
|
||||
return "settings left unchanged"
|
||||
}
|
||||
newEnabled := settings.Enabled
|
||||
previousEnabled := l.state.settings.Enabled
|
||||
l.state.settings = settings
|
||||
l.state.settingsMu.Unlock()
|
||||
// Either restart or set changed status
|
||||
switch {
|
||||
case !newEnabled && !previousEnabled:
|
||||
case newEnabled && previousEnabled:
|
||||
_, _ = l.SetStatus(ctx, constants.Stopped)
|
||||
_, _ = l.SetStatus(ctx, constants.Running)
|
||||
case newEnabled && !previousEnabled:
|
||||
_, _ = l.SetStatus(ctx, constants.Running)
|
||||
case !newEnabled && previousEnabled:
|
||||
_, _ = l.SetStatus(ctx, constants.Stopped)
|
||||
}
|
||||
return "settings updated"
|
||||
}
|
||||
@@ -1,14 +1,9 @@
|
||||
package httpproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sync"
|
||||
|
||||
"github.com/qdm12/gluetun/internal/configuration"
|
||||
"github.com/qdm12/gluetun/internal/constants"
|
||||
"github.com/qdm12/gluetun/internal/models"
|
||||
)
|
||||
|
||||
@@ -18,101 +13,3 @@ type state struct {
|
||||
statusMu sync.RWMutex
|
||||
settingsMu 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 configuration.HTTPProxy) {
|
||||
l.state.settingsMu.RLock()
|
||||
defer l.state.settingsMu.RUnlock()
|
||||
return l.state.settings
|
||||
}
|
||||
|
||||
func (l *looper) SetSettings(ctx context.Context, settings configuration.HTTPProxy) (
|
||||
outcome string) {
|
||||
l.state.settingsMu.Lock()
|
||||
settingsUnchanged := reflect.DeepEqual(settings, l.state.settings)
|
||||
if settingsUnchanged {
|
||||
l.state.settingsMu.Unlock()
|
||||
return "settings left unchanged"
|
||||
}
|
||||
newEnabled := settings.Enabled
|
||||
previousEnabled := l.state.settings.Enabled
|
||||
l.state.settings = settings
|
||||
l.state.settingsMu.Unlock()
|
||||
// Either restart or set changed status
|
||||
switch {
|
||||
case !newEnabled && !previousEnabled:
|
||||
case newEnabled && previousEnabled:
|
||||
_, _ = l.SetStatus(ctx, constants.Stopped)
|
||||
_, _ = l.SetStatus(ctx, constants.Running)
|
||||
case newEnabled && !previousEnabled:
|
||||
_, _ = l.SetStatus(ctx, constants.Running)
|
||||
case !newEnabled && previousEnabled:
|
||||
_, _ = l.SetStatus(ctx, constants.Stopped)
|
||||
}
|
||||
return "settings updated"
|
||||
}
|
||||
|
||||
76
internal/httpproxy/status.go
Normal file
76
internal/httpproxy/status.go
Normal file
@@ -0,0 +1,76 @@
|
||||
package httpproxy
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"github.com/qdm12/gluetun/internal/constants"
|
||||
"github.com/qdm12/gluetun/internal/models"
|
||||
)
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user