120 lines
2.6 KiB
Go
120 lines
2.6 KiB
Go
package daemon
|
|
|
|
import (
|
|
"context"
|
|
"net/url"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
|
|
"github.com/volker-raschek/flucky/pkg/config"
|
|
"github.com/volker-raschek/flucky/pkg/repository/db"
|
|
"github.com/volker-raschek/flucky/pkg/sensor"
|
|
"github.com/volker-raschek/flucky/pkg/types"
|
|
"github.com/volker-raschek/go-logger/pkg/logger"
|
|
)
|
|
|
|
func Start(cnf *config.Config, flogger logger.Logger) error {
|
|
|
|
sensors := make([]sensor.Sensor, 0)
|
|
for _, cnfSensor := range cnf.Sensors {
|
|
if !cnfSensor.Enabled {
|
|
continue
|
|
}
|
|
|
|
sensor, err := sensor.New(cnfSensor)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
sensors = append(sensors, sensor)
|
|
}
|
|
|
|
measuredValueChannel := make(chan *types.MeasuredValue, 0)
|
|
|
|
// load storage endpoint
|
|
storageEndpointURL, err := url.Parse(cnf.StorageEndpoint)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
backend, err := db.New(storageEndpointURL, flogger)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
interruptChannel := make(chan os.Signal, 1)
|
|
signal.Notify(interruptChannel, os.Kill, syscall.SIGTERM)
|
|
|
|
// Collection
|
|
parentCtx := context.Background()
|
|
|
|
// Insert device if not exist
|
|
device, _ := backend.SelectDevice(parentCtx, cnf.Device.ID)
|
|
if device == nil {
|
|
if err := backend.InsertDevices(parentCtx, cnf.Device); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// Insert sensors if not exist
|
|
for _, cnfSensor := range cnf.Sensors {
|
|
sensor, _ := backend.SelectSensor(parentCtx, cnfSensor.ID)
|
|
if sensor == nil {
|
|
if err := backend.InsertSensors(parentCtx, cnfSensor); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
ctx, cancel := context.WithCancel(parentCtx)
|
|
|
|
for _, s := range sensors {
|
|
go func(sensor sensor.Sensor) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-sensor.GetTicker().C:
|
|
measuredValues, err := sensor.Read()
|
|
if err != nil {
|
|
flogger.Error("%v", err)
|
|
continue
|
|
}
|
|
for _, measuredValue := range measuredValues {
|
|
measuredValueChannel <- measuredValue
|
|
}
|
|
}
|
|
}
|
|
}(s)
|
|
}
|
|
|
|
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")
|
|
err := backend.InsertMeasuredValues(ctx, measuredValues...)
|
|
if err != nil {
|
|
flogger.Error("%v", err)
|
|
}
|
|
measuredValues = make([]*types.MeasuredValue, 0, 10)
|
|
}
|
|
|
|
case <-interruptChannel:
|
|
cancel()
|
|
close(measuredValueChannel)
|
|
|
|
err := backend.InsertMeasuredValues(ctx, measuredValues...)
|
|
if err != nil {
|
|
flogger.Error("%v", err)
|
|
}
|
|
|
|
break
|
|
}
|
|
}
|
|
}
|