PKGBUILD/pkg/daemon/daemon.go

package daemon

import (
	"context"
	"fmt"
	"net/url"
	"os"
	"os/signal"

	"git.cryptic.systems/volker.raschek/flucky/pkg/config"
	"git.cryptic.systems/volker.raschek/flucky/pkg/repository"
	"git.cryptic.systems/volker.raschek/flucky/pkg/sensor"
	"git.cryptic.systems/volker.raschek/flucky/pkg/types"
	"git.cryptic.systems/volker.raschek/go-logger"
)

func Start(cnf *config.Config, flogger logger.Logger) error {

	// load data source name (dsn)
	dsnURL, err := url.Parse(cnf.DSN)
	if err != nil {
		return err
	}

	repo, err := repository.New(dsnURL, flogger)
	if err != nil {
		return err
	}

	// Add
	repoDevice, err := repo.GetDevice(cnf.DeviceID)
	switch {
	case err != nil:
		return err
	case repoDevice == nil:

		hostname, err := os.Hostname()
		if err != nil {
			return err
		}

		err = repo.AddDevices(&types.Device{
			ID:   cnf.DeviceID,
			Name: hostname,
		})
		if err != nil {
			return err
		}

		repoDevice, err = repo.GetDevice(cnf.DeviceID)
		if err != nil {
			return err
		}
	}

	repoSensors, err := repo.GetSensorsByDeviceID(repoDevice.ID)
	switch {
	case err != nil:
		return err
	case repoSensors == nil, len(repoSensors) <= 0:
		return fmt.Errorf("No sensors found")
	}

	sensors := make([]sensor.Sensor, 0)
	for _, repoSensor := range repoSensors {
		if !repoSensor.Enabled || repoSensor.DeviceID != repoDevice.ID {
			continue
		}

		flogger.Debug("Found sensor %v", repoSensor.Name)

		sensor, err := sensor.New(repoSensor)
		if err != nil {
			return err
		}
		sensors = append(sensors, sensor)
	}

	interruptChannel := make(chan os.Signal, 1)
	signal.Notify(interruptChannel, os.Interrupt, os.Kill)

	// Collection
	parentCtx := context.Background()
	ctx, cancel := context.WithCancel(parentCtx)

	measuredValueChannel, errorChannel := sensor.ReadTickingPipeline(ctx, sensors...)

	go func() {
		for {
			select {
			case <-ctx.Done():
				return
			case err, open := <-errorChannel:
				if !open {
					return
				}
				if err != nil {
					flogger.Error("%v", err)
				}
			}
		}
	}()

	measuredValues := make([]*types.MeasuredValue, 0, 10)
	for {
		select {
		case measuredValue := <-measuredValueChannel:
			flogger.Debug("%v\t%v\t%v", measuredValue.ID, measuredValue.ValueType, measuredValue.Value)
			measuredValues = append(measuredValues, measuredValue)

			if cap(measuredValues) == len(measuredValues) {
				flogger.Debug("Flush cache with %v values", len(measuredValues))
				err := repo.AddMeasuredValues(measuredValues...)
				if err != nil {
					flogger.Error("%v", err)
				}
				measuredValues = make([]*types.MeasuredValue, 0, 10)
			}

		case signal := <-interruptChannel:
			cancel()

			flogger.Info("Stopping daemon: Received process signal %v", signal.String())

			flogger.Debug("Flush cache with %v remaining values", len(measuredValues))

			err := repo.AddMeasuredValues(measuredValues...)
			if err != nil {
				flogger.Error("%v", err)
			}

			flogger.Debug("Close repository")
			err = repo.Close()
			if err != nil {
				flogger.Error("%v", err)
			}

			return nil
		}
	}
}