fix(pkg/sensors): Use channel of data type []measuredValues instead of measuredValues

pkg/daemon/daemon.go

@@ -28,7 +28,7 @@ func Start(cnf *config.Configuration, cleanCacheInterval time.Duration, compress
 	signal.Notify(interrupt, os.Interrupt, os.Kill, syscall.SIGTERM)
 
 	errorChannel := make(chan error, 0)
-	measuredValueChannel := make(chan types.MeasuredValue, 0)
+	measuredValuesChannel := make(chan []types.MeasuredValue, 0)
 
 	ctx := context.Background()
 	childContext, cancel := context.WithCancel(ctx)
@@ -37,7 +37,7 @@ func Start(cnf *config.Configuration, cleanCacheInterval time.Duration, compress
 
 	measuredValuesCache := make([]types.MeasuredValue, 0)
 
-	go sensor.ReadContinuously(childContext, cnf.GetTemperatureSensors(config.ENABLED), measuredValueChannel, errorChannel)
+	go sensor.ReadContinuously(childContext, cnf.GetTemperatureSensors(config.ENABLED), measuredValuesChannel, errorChannel)
 
 	rgbLEDs := cnf.GetRGBLEDs(config.ENABLED)
 
@@ -79,8 +79,8 @@ func Start(cnf *config.Configuration, cleanCacheInterval time.Duration, compress
 			}
 			measuredValuesCache = make([]types.MeasuredValue, 0)
 
-		case measuredValue, _ := <-measuredValueChannel:
+		case measuredValues, _ := <-measuredValuesChannel:
-			measuredValuesCache = append(measuredValuesCache, measuredValue)
+			measuredValuesCache = append(measuredValuesCache, measuredValues...)
 
 		case killSignal := <-interrupt:
 			logger.Warn("Daemon was interruped by system signal %v\n", killSignal)

pkg/internal/collect/measuredValues.go

@@ -4,13 +4,13 @@ import (
 	"github.com/go-flucky/flucky/pkg/types"
 )
 
-func MeasuredValues(measuredValueChannel <-chan types.MeasuredValue) []types.MeasuredValue {
+func MeasuredValues(measuredValuesChannel <-chan []types.MeasuredValue) []types.MeasuredValue {
 	cachedMeasuredValues := make([]types.MeasuredValue, 0)
 	for {
 		select {
-		case measuredValue, more := <-measuredValueChannel:
+		case measuredValues, more := <-measuredValuesChannel:
 			if more {
-				cachedMeasuredValues = append(cachedMeasuredValues, measuredValue)
+				cachedMeasuredValues = append(cachedMeasuredValues, measuredValues...)
 				continue
 			}
 		default:

pkg/sensor/dht11.go

@@ -66,7 +66,7 @@ func (s *DHT11) Read() ([]types.MeasuredValue, error) {
 
 // ReadChannel reads the measured values from the sensor and writes them to a
 // channel.
-func (s *DHT11) ReadChannel(measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup) {
+func (s *DHT11) ReadChannel(measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup) {
 	if wg != nil {
 		defer wg.Done()
 	}
@@ -77,21 +77,20 @@ func (s *DHT11) ReadChannel(measuredValueChannel chan<- types.MeasuredValue, err
 		return
 	}
 
-	for _, measuredValue := range measuredValues {
+	measuredValuesChannel <- measuredValues
-		measuredValueChannel <- measuredValue
+
-	}
 }
 
 // ReadContinously reads the measured values continously from the sensor and
 // writes them to a channel.
-func (s *DHT11) ReadContinously(ctx context.Context, measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error) {
+func (s *DHT11) ReadContinously(ctx context.Context, measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error) {
 	for {
 		select {
 		case <-ctx.Done():
 			errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
 			return
 		default:
-			s.ReadChannel(measuredValueChannel, errorChannel, nil)
+			s.ReadChannel(measuredValuesChannel, errorChannel, nil)
 		}
 	}
 }

pkg/sensor/dht22.go

@@ -66,7 +66,7 @@ func (s *DHT22) Read() ([]types.MeasuredValue, error) {
 
 // ReadChannel reads the measured values from the sensor and writes them to a
 // channel.
-func (s *DHT22) ReadChannel(measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup) {
+func (s *DHT22) ReadChannel(measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup) {
 	if wg != nil {
 		defer wg.Done()
 	}
@@ -77,21 +77,20 @@ func (s *DHT22) ReadChannel(measuredValueChannel chan<- types.MeasuredValue, err
 		return
 	}
 
-	for _, measuredValue := range measuredValues {
+	measuredValuesChannel <- measuredValues
-		measuredValueChannel <- measuredValue
+
-	}
 }
 
 // ReadContinously reads the measured values continously from the sensor and
 // writes them to a channel.
-func (s *DHT22) ReadContinously(ctx context.Context, measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error) {
+func (s *DHT22) ReadContinously(ctx context.Context, measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error) {
 	for {
 		select {
 		case <-ctx.Done():
 			errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
 			return
 		default:
-			s.ReadChannel(measuredValueChannel, errorChannel, nil)
+			s.ReadChannel(measuredValuesChannel, errorChannel, nil)
 		}
 	}
 }

pkg/sensor/ds18b20.go

@@ -66,7 +66,7 @@ func (s *DS18B20) Read() ([]types.MeasuredValue, error) {
 
 // ReadChannel reads the measured values from the sensor and writes them to a
 // channel.
-func (s *DS18B20) ReadChannel(measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup) {
+func (s *DS18B20) ReadChannel(measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup) {
 	if wg != nil {
 		defer wg.Done()
 	}
@@ -77,22 +77,20 @@ func (s *DS18B20) ReadChannel(measuredValueChannel chan<- types.MeasuredValue, e
 		return
 	}
 
-	for _, measuredValue := range measuredValues {
+	measuredValuesChannel <- measuredValues
-		measuredValueChannel <- measuredValue
-	}
 
 }
 
 // ReadContinously reads the measured values continously from the sensor and
 // writes them to a channel.
-func (s *DS18B20) ReadContinously(ctx context.Context, measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error) {
+func (s *DS18B20) ReadContinously(ctx context.Context, measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error) {
 	for {
 		select {
 		case <-ctx.Done():
 			errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
 			return
 		default:
-			s.ReadChannel(measuredValueChannel, errorChannel, nil)
+			s.ReadChannel(measuredValuesChannel, errorChannel, nil)
 		}
 	}
 }

pkg/sensor/interfaces.go

@@ -10,6 +10,6 @@ import (
 type Sensor interface {
 	GetSensorModel() types.SensorModel
 	Read() ([]types.MeasuredValue, error)
-	ReadChannel(measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup)
+	ReadChannel(measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error, wg *sync.WaitGroup)
-	ReadContinously(ctx context.Context, measuredValueChannel chan<- types.MeasuredValue, errorChannel chan<- error)
+	ReadContinously(ctx context.Context, measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error)
 }

pkg/sensor/sensor.go

@@ -5,45 +5,54 @@ import (
 	"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) {
 
-	// TODO: Execute Read with go function
+	measuredValuesChannel := make(chan []types.MeasuredValue, len(sensors))
-	cachesMeasuredValues := make([]types.MeasuredValue, 0)
+	errorChannel := make(chan error, len(sensors))
 
-	for _, sensor := range sensors {
+	ReadChannel(ctx, sensors, measuredValuesChannel, errorChannel)
-		measuredValues, err := sensor.Read()
+
-		if err != nil {
+	errors := collect.Errors(errorChannel)
-			return nil, err
+	if len(errors) > 0 {
-		}
+		return nil, prittyprint.FormatErrors(errors)
-		cachesMeasuredValues = append(cachesMeasuredValues, measuredValues...)
 	}
 
-	return cachesMeasuredValues, nil
+	measuredValues := collect.MeasuredValues(measuredValuesChannel)
+
+	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 *sync.WaitGroup) {
+func ReadChannel(ctx context.Context, sensors []Sensor, measuredValuesChannel chan<- []types.MeasuredValue, errorChannel chan<- error) {
-	// TODO: Execute ReadCallel with go function if wg is available
+
+	wg := new(sync.WaitGroup)
+	wg.Add(len(sensors))
+
 	for _, sensor := range sensors {
-		sensor.ReadChannel(measuredValueChannel, errorChannel, wg)
+		go sensor.ReadChannel(measuredValuesChannel, 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) {
+func ReadContinuously(ctx context.Context, sensors []Sensor, measuredValuesChannel 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, nil)
+			ReadChannel(ctx, sensors, measuredValuesChannel, errorChannel)
 		}
 	}
 }