gluetun/internal/portforward/loop.go

package portforward

import (
	"context"
	"fmt"
	"net/http"
	"sync"

	"github.com/qdm12/gluetun/internal/configuration/settings"
	"github.com/qdm12/gluetun/internal/portforward/service"
)

type Loop struct {
	// State
	settings      service.Settings
	settingsMutex sync.RWMutex
	service       Service
	// Fixed injected objets
	routing     Routing
	client      *http.Client
	portAllower PortAllower
	logger      Logger
	// Fixed parameters
	uid, gid int
	// Internal channels and locks
	// runCtx is used to detect when the loop has exited
	// when performing an update
	runCtx        context.Context //nolint:containedctx
	runCancel     context.CancelFunc
	runDone       <-chan struct{}
	updateTrigger chan<- service.Settings
	updatedResult <-chan error
}

func NewLoop(settings settings.PortForwarding, routing Routing,
	client *http.Client, portAllower PortAllower,
	logger Logger, uid, gid int) *Loop {
	return &Loop{
		settings: service.Settings{
			UserSettings: settings,
		},
		routing:     routing,
		client:      client,
		portAllower: portAllower,
		logger:      logger,
		uid:         uid,
		gid:         gid,
	}
}

func (l *Loop) Start(_ context.Context) (runError <-chan error, _ error) {
	l.runCtx, l.runCancel = context.WithCancel(context.Background())
	runDone := make(chan struct{})
	l.runDone = runDone

	updateTrigger := make(chan service.Settings)
	l.updateTrigger = updateTrigger
	updateResult := make(chan error)
	l.updatedResult = updateResult
	runErrorCh := make(chan error)

	go l.run(l.runCtx, runDone, runErrorCh,
		l.settings, updateTrigger, updateResult)

	return runErrorCh, nil
}

func (l *Loop) run(runCtx context.Context, runDone chan<- struct{},
	runErrorCh chan<- error, initialSettings service.Settings,
	updateTrigger <-chan service.Settings, updateResult chan<- error) {
	defer close(runDone)

	settings := initialSettings
	var serviceRunError <-chan error
	for {
		updateReceived := false
		select {
		case <-runCtx.Done():
			// Stop call takes care of stopping the service
			return
		case partialUpdate := <-updateTrigger:
			updatedSettings, err := settings.UpdateWith(partialUpdate)
			if err != nil {
				updateResult <- err
				continue
			}
			settings = updatedSettings
			updateReceived = true
		case err := <-serviceRunError:
			l.logger.Error(err.Error())
		}

		firstRun := l.service == nil
		if !firstRun {
			err := l.service.Stop()
			if err != nil {
				runErrorCh <- fmt.Errorf("stopping previous service: %w", err)
				return
			}
		}

		l.service = service.New(settings, l.routing, l.client,
			l.portAllower, l.logger, l.uid, l.gid)

		var err error
		serviceRunError, err = l.service.Start(runCtx)
		if updateReceived {
			// Signal to the Update call that the service has started
			// and if it failed to start.
			updateResult <- err
		}
		if err != nil {
			if runCtx.Err() == nil { // crashed but NOT stopped
				runErrorCh <- fmt.Errorf("starting new service: %w", err)
			}
			return
		}

		// Service is created and started successfully, so update
		// the settings for external calls such as GetSettings.
		l.settingsMutex.Lock()
		l.settings = settings
		l.settingsMutex.Unlock()
	}
}

func (l *Loop) UpdateWith(partialUpdate service.Settings) (err error) {
	select {
	case l.updateTrigger <- partialUpdate:
		select {
		case err = <-l.updatedResult:
			return err
		case <-l.runCtx.Done():
			return l.runCtx.Err()
		}
	case <-l.runCtx.Done():
		// loop has been stopped, no update can be done
		return l.runCtx.Err()
	}
}

func (l *Loop) Stop() (err error) {
	l.runCancel()
	<-l.runDone

	if l.service != nil {
		return l.service.Stop()
	}
	return nil
}

func (l *Loop) GetPortForwarded() (port uint16) {
	if l.service == nil {
		return 0
	}
	return l.service.GetPortForwarded()
}