fix(pkg/sensor): use context to terminate go routines
This commit is contained in:
@ -1,7 +1,9 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-flucky/flucky/pkg/types"
|
||||
@ -52,6 +54,33 @@ func (s *DHT11) ReadHumidity() (*types.Humidity, error) {
|
||||
return humidity, nil
|
||||
}
|
||||
|
||||
// ReadHumidityWriteIntoChannel and write values into a channel
|
||||
func (s *DHT11) ReadHumidityWriteIntoChannel(humidityChannel chan<- *types.Humidity, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
if wg != nil {
|
||||
defer wg.Done()
|
||||
}
|
||||
|
||||
humidity, err := s.ReadHumidity()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
humidityChannel <- humidity
|
||||
}
|
||||
|
||||
// ReadHumidityContinously into a channel until context closed
|
||||
func (s *DHT11) ReadHumidityContinously(ctx context.Context, humidityChannel chan<- *types.Humidity, errorChannel chan<- error) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
|
||||
return
|
||||
default:
|
||||
s.ReadHumidityWriteIntoChannel(humidityChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ReadTemperature measure the temperature
|
||||
func (s *DHT11) ReadTemperature() (*types.Temperature, error) {
|
||||
err := dht.HostInit()
|
||||
@ -84,3 +113,30 @@ func (s *DHT11) ReadTemperature() (*types.Temperature, error) {
|
||||
|
||||
return temperature, nil
|
||||
}
|
||||
|
||||
// ReadTemperatureWriteIntoChannel and write values into a channel
|
||||
func (s *DHT11) ReadTemperatureWriteIntoChannel(temperatureChannel chan<- *types.Temperature, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
if wg != nil {
|
||||
defer wg.Done()
|
||||
}
|
||||
|
||||
temperature, err := s.ReadTemperature()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
temperatureChannel <- temperature
|
||||
}
|
||||
|
||||
// ReadTemperatureContinously into a channel until context closed
|
||||
func (s *DHT11) ReadTemperatureContinously(ctx context.Context, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
|
||||
return
|
||||
default:
|
||||
s.ReadTemperatureWriteIntoChannel(temperatureChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -1,7 +1,9 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-flucky/flucky/pkg/types"
|
||||
@ -52,6 +54,33 @@ func (s *DHT22) ReadHumidity() (*types.Humidity, error) {
|
||||
return humidity, nil
|
||||
}
|
||||
|
||||
// ReadHumidityWriteIntoChannel and write values into a channel
|
||||
func (s *DHT22) ReadHumidityWriteIntoChannel(humidityChannel chan<- *types.Humidity, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
if wg != nil {
|
||||
defer wg.Done()
|
||||
}
|
||||
|
||||
humidity, err := s.ReadHumidity()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
humidityChannel <- humidity
|
||||
}
|
||||
|
||||
// ReadHumidityContinously into a channel until context closed
|
||||
func (s *DHT22) ReadHumidityContinously(ctx context.Context, humidityChannel chan<- *types.Humidity, errorChannel chan<- error) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
|
||||
return
|
||||
default:
|
||||
s.ReadHumidityWriteIntoChannel(humidityChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ReadTemperature measure the temperature
|
||||
func (s *DHT22) ReadTemperature() (*types.Temperature, error) {
|
||||
err := dht.HostInit()
|
||||
@ -84,3 +113,30 @@ func (s *DHT22) ReadTemperature() (*types.Temperature, error) {
|
||||
|
||||
return temperature, nil
|
||||
}
|
||||
|
||||
// ReadTemperatureWriteIntoChannel and write values into a channel
|
||||
func (s *DHT22) ReadTemperatureWriteIntoChannel(temperatureChannel chan<- *types.Temperature, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
if wg != nil {
|
||||
defer wg.Done()
|
||||
}
|
||||
|
||||
temperature, err := s.ReadTemperature()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
temperatureChannel <- temperature
|
||||
}
|
||||
|
||||
// ReadTemperatureContinously into a channel until context closed
|
||||
func (s *DHT22) ReadTemperatureContinously(ctx context.Context, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
|
||||
return
|
||||
default:
|
||||
s.ReadTemperatureWriteIntoChannel(temperatureChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -1,12 +1,17 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-flucky/flucky/pkg/types"
|
||||
uuid "github.com/satori/go.uuid"
|
||||
"github.com/yryz/ds18b20"
|
||||
)
|
||||
|
||||
// DS18B20 is a sensor to measure humidity and temperature.
|
||||
@ -27,14 +32,26 @@ func (s *DS18B20) GetSensor() *types.Sensor {
|
||||
// ReadTemperature measure the temperature
|
||||
func (s *DS18B20) ReadTemperature() (*types.Temperature, error) {
|
||||
|
||||
t, err := ds18b20.Temperature(*s.WireID)
|
||||
data, err := ioutil.ReadFile(filepath.Join("/sys/bus/w1/devices", *s.WireID, "/w1_slave"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("Can not read from Sensor %v (UUID: %v, Wire-ID: %v): %v", s.SensorName, s.SensorID, s.WireID, err)
|
||||
return nil, fmt.Errorf("Can not read data from sensor %v", s.SensorName)
|
||||
}
|
||||
|
||||
raw := string(data)
|
||||
|
||||
i := strings.LastIndex(raw, "t=")
|
||||
if i == -1 {
|
||||
return nil, ErrReadSensor
|
||||
}
|
||||
|
||||
celsius, err := strconv.ParseFloat(raw[i+2:len(raw)-1], 64)
|
||||
if err != nil {
|
||||
return nil, ErrParseData
|
||||
}
|
||||
|
||||
temperature := &types.Temperature{
|
||||
TemperatureID: uuid.NewV4().String(),
|
||||
TemperatureValue: t,
|
||||
TemperatureValue: celsius / 1000,
|
||||
TemperatureFromDate: time.Now(),
|
||||
TemperatureTillDate: time.Now(),
|
||||
SensorID: s.SensorID,
|
||||
@ -43,3 +60,30 @@ func (s *DS18B20) ReadTemperature() (*types.Temperature, error) {
|
||||
return temperature, nil
|
||||
|
||||
}
|
||||
|
||||
// ReadTemperatureWriteIntoChannel and write values into a channel
|
||||
func (s *DS18B20) ReadTemperatureWriteIntoChannel(temperatureChannel chan<- *types.Temperature, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
if wg != nil {
|
||||
defer wg.Done()
|
||||
}
|
||||
|
||||
temperature, err := s.ReadTemperature()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
temperatureChannel <- temperature
|
||||
}
|
||||
|
||||
// ReadTemperatureContinously into a channel until context closed
|
||||
func (s *DS18B20) ReadTemperatureContinously(ctx context.Context, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("%v: Context closed: %v", s.SensorName, ctx.Err())
|
||||
return
|
||||
default:
|
||||
s.ReadTemperatureWriteIntoChannel(temperatureChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
8
pkg/sensor/error.go
Normal file
8
pkg/sensor/error.go
Normal file
@ -0,0 +1,8 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
"errors"
|
||||
)
|
||||
|
||||
var ErrParseData = errors.New("Can not parse data")
|
||||
var ErrReadSensor = errors.New("Can not read data from Sensor")
|
@ -1,15 +1,24 @@
|
||||
package sensor
|
||||
|
||||
import "github.com/go-flucky/flucky/pkg/types"
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/go-flucky/flucky/pkg/types"
|
||||
)
|
||||
|
||||
// HumiditySensor is a interface to describe required functions to measure humidities
|
||||
type HumiditySensor interface {
|
||||
GetSensorModel() types.SensorModel
|
||||
ReadHumidity() (*types.Humidity, error)
|
||||
ReadHumidityWriteIntoChannel(humidityChannel chan<- *types.Humidity, errorChannel chan<- error, wg *sync.WaitGroup)
|
||||
ReadHumidityContinously(ctx context.Context, humidityChannel chan<- *types.Humidity, errorChannel chan<- error)
|
||||
}
|
||||
|
||||
// TemperatureSensor is a interface to describe required functions to measure temperatures
|
||||
type TemperatureSensor interface {
|
||||
GetSensorModel() types.SensorModel
|
||||
ReadTemperature() (*types.Temperature, error)
|
||||
ReadTemperatureWriteIntoChannel(temperatureChannel chan<- *types.Temperature, errorChannel chan<- error, wg *sync.WaitGroup)
|
||||
ReadTemperatureContinously(ctx context.Context, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error)
|
||||
}
|
||||
|
@ -1,6 +1,9 @@
|
||||
package sensor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"sync"
|
||||
|
||||
"github.com/go-flucky/flucky/pkg/internal/collect"
|
||||
@ -9,81 +12,102 @@ import (
|
||||
)
|
||||
|
||||
// ReadHumidities returns a list of measured humidities by humidity sensors
|
||||
func ReadHumidities(humiditySensors []HumiditySensor) ([]*types.Humidity, error) {
|
||||
errorChannel := make(chan error, len(humiditySensors))
|
||||
func ReadHumidities(humiditySensors []HumiditySensor, round float64) ([]*types.Humidity, error) {
|
||||
humidityChannel := make(chan *types.Humidity, len(humiditySensors))
|
||||
ReadHumiditiesIntoChannel(humiditySensors, humidityChannel, errorChannel)
|
||||
errorChannel := make(chan error, len(humiditySensors))
|
||||
|
||||
errorList := collect.Errors(errorChannel)
|
||||
if len(errorList) != 0 {
|
||||
return nil, prittyprint.FormatErrors(errorList)
|
||||
}
|
||||
|
||||
humidityList := collect.Humidities(humidityChannel)
|
||||
return humidityList, nil
|
||||
}
|
||||
|
||||
// ReadHumiditiesIntoChannel reads the humidity values of humidity sensors and writes them into a channel
|
||||
func ReadHumiditiesIntoChannel(humiditySensors []HumiditySensor, humidityChannel chan<- *types.Humidity, errorChannel chan<- error) {
|
||||
wg := new(sync.WaitGroup)
|
||||
wg.Add(len(humiditySensors))
|
||||
|
||||
for _, humiditySensor := range humiditySensors {
|
||||
go func(hs HumiditySensor) {
|
||||
defer wg.Done()
|
||||
humidity, err := hs.ReadHumidity()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
humidityChannel <- humidity
|
||||
}(humiditySensor)
|
||||
go humiditySensor.ReadHumidityWriteIntoChannel(humidityChannel, errorChannel, wg)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
errors := collect.Errors(errorChannel)
|
||||
if len(errors) > 0 {
|
||||
return nil, prittyprint.FormatErrors(errors)
|
||||
}
|
||||
|
||||
humidities := collect.Humidities(humidityChannel)
|
||||
|
||||
if round != 0 {
|
||||
for _, humidity := range humidities {
|
||||
humidity.HumidityValue = math.Round(humidity.HumidityValue/round) * round
|
||||
}
|
||||
}
|
||||
|
||||
return humidities, nil
|
||||
}
|
||||
|
||||
// ReadHumiditiesWriteIntoChannel reads the humidity values of humidity sensors and writes them into a channel
|
||||
func ReadHumiditiesWriteIntoChannel(ctx context.Context, humiditySensors []HumiditySensor, humidityChannel chan<- *types.Humidity, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
for _, humiditySensor := range humiditySensors {
|
||||
humiditySensor.ReadHumidityWriteIntoChannel(humidityChannel, errorChannel, wg)
|
||||
}
|
||||
}
|
||||
|
||||
// ReadHumiditiesContinuously reads the humidity values of humidity sensors continuously and writes them into a channel
|
||||
func ReadHumiditiesContinuously(humiditySensors []HumiditySensor, humidityChannel chan<- *types.Humidity, errorChannel chan<- error) {
|
||||
func ReadHumiditiesContinuously(ctx context.Context, humiditySensors []HumiditySensor, humidityChannel chan<- *types.Humidity, errorChannel chan<- error) {
|
||||
for {
|
||||
ReadHumiditiesIntoChannel(humiditySensors, humidityChannel, errorChannel)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("Context closed: %v", ctx.Err())
|
||||
return
|
||||
default:
|
||||
ReadHumiditiesWriteIntoChannel(ctx, humiditySensors, humidityChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ReadTemperatures returns a list of measured temperatures by temperature sensors
|
||||
func ReadTemperatures(temperatureSensors []TemperatureSensor) ([]*types.Temperature, error) {
|
||||
errorChannel := make(chan error, len(temperatureSensors))
|
||||
func ReadTemperatures(temperatureSensors []TemperatureSensor, round float64) ([]*types.Temperature, error) {
|
||||
temperatureChannel := make(chan *types.Temperature, len(temperatureSensors))
|
||||
ReadTemperaturesIntoChannel(temperatureSensors, temperatureChannel, errorChannel)
|
||||
errorChannel := make(chan error, len(temperatureSensors))
|
||||
|
||||
errorList := collect.Errors(errorChannel)
|
||||
if len(errorList) != 0 {
|
||||
return nil, prittyprint.FormatErrors(errorList)
|
||||
}
|
||||
|
||||
temperatureList := collect.Temperatures(temperatureChannel)
|
||||
return temperatureList, nil
|
||||
}
|
||||
|
||||
// ReadTemperaturesIntoChannel reads the temperature values of temperature sensors and writes them into a channel
|
||||
func ReadTemperaturesIntoChannel(temperatureSensors []TemperatureSensor, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error) {
|
||||
wg := new(sync.WaitGroup)
|
||||
wg.Add(len(temperatureSensors))
|
||||
|
||||
for _, temperatureSensor := range temperatureSensors {
|
||||
go func(ts TemperatureSensor) {
|
||||
defer wg.Done()
|
||||
temperature, err := ts.ReadTemperature()
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
temperatureChannel <- temperature
|
||||
}(temperatureSensor)
|
||||
go temperatureSensor.ReadTemperatureWriteIntoChannel(temperatureChannel, errorChannel, wg)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
errors := collect.Errors(errorChannel)
|
||||
if len(errors) > 0 {
|
||||
return nil, prittyprint.FormatErrors(errors)
|
||||
}
|
||||
|
||||
temperatures := collect.Temperatures(temperatureChannel)
|
||||
|
||||
if round != 0 {
|
||||
for _, temperature := range temperatures {
|
||||
temperature.TemperatureValue = math.Round(temperature.TemperatureValue/round) * round
|
||||
}
|
||||
}
|
||||
|
||||
return temperatures, nil
|
||||
}
|
||||
|
||||
// ReadTemperaturesWriteIntoChannel reads the temperature values of temperature sensors and writes them into a channel
|
||||
func ReadTemperaturesWriteIntoChannel(ctx context.Context, temperatureSensors []TemperatureSensor, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error, wg *sync.WaitGroup) {
|
||||
for _, temperatureSensor := range temperatureSensors {
|
||||
temperatureSensor.ReadTemperatureWriteIntoChannel(temperatureChannel, errorChannel, wg)
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// ReadTemperaturesContinuously reads the temperature values of temperature sensors continuously and writes them into a chann
|
||||
func ReadTemperaturesContinuously(temperatureSensors []TemperatureSensor, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error) {
|
||||
func ReadTemperaturesContinuously(ctx context.Context, temperatureSensors []TemperatureSensor, temperatureChannel chan<- *types.Temperature, errorChannel chan<- error) {
|
||||
for {
|
||||
ReadTemperaturesIntoChannel(temperatureSensors, temperatureChannel, errorChannel)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errorChannel <- fmt.Errorf("Context closed: %v", ctx.Err())
|
||||
return
|
||||
default:
|
||||
ReadTemperaturesWriteIntoChannel(ctx, temperatureSensors, temperatureChannel, errorChannel, nil)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
Reference in New Issue
Block a user