Markus Pesch
dbef4f8241
changes: - Only one storage endpoint can be defined. This consists of a URL which can be used to specify whether the data is to be stored in a file or in a database.
59 lines
1.5 KiB
Go
59 lines
1.5 KiB
Go
package sensor
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sync"
|
|
|
|
"github.com/go-flucky/flucky/pkg/internal/collect"
|
|
"github.com/go-flucky/flucky/pkg/internal/prittyprint"
|
|
|
|
"github.com/go-flucky/flucky/pkg/types"
|
|
)
|
|
|
|
// Read measured values from sensors
|
|
func Read(ctx context.Context, sensors []Sensor) ([]*types.MeasuredValue, error) {
|
|
|
|
measuredValueChannel := make(chan *types.MeasuredValue, len(sensors))
|
|
errorChannel := make(chan error, len(sensors))
|
|
|
|
ReadChannel(ctx, sensors, measuredValueChannel, errorChannel)
|
|
|
|
errors := collect.Errors(errorChannel)
|
|
if len(errors) > 0 {
|
|
return nil, prittyprint.FormatErrors(errors)
|
|
}
|
|
|
|
measuredValues := collect.MeasuredValues(measuredValueChannel)
|
|
|
|
return measuredValues, nil
|
|
}
|
|
|
|
// ReadChannel reads the measured values from sensors and writes them to a
|
|
// channel.
|
|
func ReadChannel(ctx context.Context, sensors []Sensor, measuredValueChannel chan<- *types.MeasuredValue, errorChannel chan<- error) {
|
|
|
|
wg := new(sync.WaitGroup)
|
|
wg.Add(len(sensors))
|
|
|
|
for _, sensor := range sensors {
|
|
go sensor.ReadChannel(measuredValueChannel, errorChannel, wg)
|
|
}
|
|
|
|
wg.Wait()
|
|
}
|
|
|
|
// ReadContinuously reads the measured values continously from sensors and writes
|
|
// them to a channel.
|
|
func ReadContinuously(ctx context.Context, sensors []Sensor, measuredValueChannel chan<- *types.MeasuredValue, errorChannel chan<- error) {
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
errorChannel <- fmt.Errorf("Context closed: %v", ctx.Err())
|
|
return
|
|
default:
|
|
ReadChannel(ctx, sensors, measuredValueChannel, errorChannel)
|
|
}
|
|
}
|
|
}
|