Stefan/evcc-io
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add util.Waiter for initial value and timeout handling (#174)

Browse source
This commit is contained in:
andig 2020-05-20 15:54:16 +02:00 • committed by GitHub
parent d93dcef709
commit 0490db10fa
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 98 additions and 104 deletions
Download patch file Download diff file Expand all files Collapse all files

47
meter/sma.go
View file

@ -2,7 +2,6 @@ package meter
import (
"fmt"
"sync"
"time"
"github.com/andig/evcc/api"
@ -10,10 +9,7 @@ import (
"github.com/andig/evcc/util"
)
const (
udpTimeout = 10 * time.Second
waitTimeout = 50 * time.Millisecond // interval when waiting for initial value
)
const udpTimeout = 10 * time.Second
// values bundles SMA readings
type values struct {
@ -27,15 +23,13 @@ type values struct {
// SMA supporting SMA Home Manager 2.0 and SMA Energy Meter 30
type SMA struct {
log *util.Logger
mux *util.Waiter
uri string
serial string
values values
powerO sma.Obis
energyO sma.Obis
updated time.Time
recv chan sma.Telegram
mux sync.Mutex
once sync.Once
}
// NewSMAFromConfig creates a SMA Meter from generic config
@ -50,9 +44,10 @@ func NewSMAFromConfig(log *util.Logger, other map[string]interface{}) api.Meter
// NewSMA creates a SMA Meter
func NewSMA(uri, serial, power, energy string) api.Meter {
log := util.NewLogger("sma ")
log := util.NewLogger("sma")
sm := &SMA{
mux: util.NewWaiter(udpTimeout, func() { log.TRACE.Println("wait for initial value") }),
log: log,
uri: uri,
serial: serial,
@ -93,17 +88,17 @@ func (sm *SMA) updateMeterValues(msg sma.Telegram) {
// use user-defined obis
if power, ok := msg.Values[sm.powerO]; ok {
sm.values.power = power
sm.updated = time.Now()
sm.mux.Update()
}
} else {
sm.values.power = msg.Values[sma.ImportPower] - msg.Values[sma.ExportPower]
sm.updated = time.Now()
sm.mux.Update()
}
if sm.energyO != "" {
if energy, ok := msg.Values[sm.energyO]; ok {
sm.values.energy = energy
sm.updated = time.Now()
sm.mux.Update()
} else {
sm.log.WARN.Println("missing obis for energy")
}
@ -111,21 +106,21 @@ func (sm *SMA) updateMeterValues(msg sma.Telegram) {
if currentL1, ok := msg.Values[sma.CurrentL1]; ok {
sm.values.currentL1 = currentL1
sm.updated = time.Now()
sm.mux.Update()
} else {
sm.log.WARN.Println("missing obis for currentL1")
}
if currentL2, ok := msg.Values[sma.CurrentL2]; ok {
sm.values.currentL2 = currentL2
sm.updated = time.Now()
sm.mux.Update()
} else {
sm.log.WARN.Println("missing obis for currentL2")
}
if currentL3, ok := msg.Values[sma.CurrentL3]; ok {
sm.values.currentL3 = currentL3
sm.updated = time.Now()
sm.mux.Update()
} else {
sm.log.WARN.Println("missing obis for currentL3")
}
@ -142,29 +137,11 @@ func (sm *SMA) receive() {
}
}
// waitForInitialValue makes sure we don't start with an error
func (sm *SMA) waitForInitialValue() {
sm.mux.Lock()
defer sm.mux.Unlock()
if sm.updated.IsZero() {
sm.log.TRACE.Print("waiting for initial value")
// wait for initial update
for sm.updated.IsZero() {
sm.mux.Unlock()
time.Sleep(waitTimeout)
sm.mux.Lock()
}
}
}
func (sm *SMA) hasValue() (values, error) {
sm.once.Do(sm.waitForInitialValue)
sm.mux.Lock()
elapsed := sm.mux.LockWithTimeout()
defer sm.mux.Unlock()
if elapsed := time.Since(sm.updated); elapsed > udpTimeout {
if elapsed > 0 {
return values{}, fmt.Errorf("recv timeout: %v", elapsed.Truncate(time.Second))
}

1
meter/sma_test.go
View file

@ -53,6 +53,7 @@ func TestSMAUpdateMeterValues(t *testing.T) {
t.Run(tt.name, func(t *testing.T) {
sm := &SMA{
log: util.NewLogger("sma "),
mux: util.NewWaiter(udpTimeout, func() {}),
}
sm.updateMeterValues(tt.messsage)

61
provider/mqtt.go
View file

@ -11,10 +11,7 @@ import (
mqtt "github.com/eclipse/paho.mqtt.golang"
)
const (
publishTimeout = 2 * time.Second
waitTimeout = 50 * time.Millisecond // polling interval when waiting for initial value
)
const publishTimeout = 2 * time.Second
// MqttClient is a paho publisher
type MqttClient struct {
@ -107,10 +104,10 @@ func (m *MqttClient) listen(topic string, callback func(string)) {
// FloatGetter creates handler for float64 from MQTT topic that returns cached value
func (m *MqttClient) FloatGetter(topic string, scale float64, timeout time.Duration) FloatGetter {
h := &msgHandler{
log: m.log,
topic: topic,
scale: scale,
timeout: timeout,
log: m.log,
mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }),
topic: topic,
scale: scale,
}
m.Listen(topic, h.Receive)
@ -120,10 +117,10 @@ func (m *MqttClient) FloatGetter(topic string, scale float64, timeout time.Durat
// IntGetter creates handler for int64 from MQTT topic that returns cached value
func (m *MqttClient) IntGetter(topic string, scale int64, timeout time.Duration) IntGetter {
h := &msgHandler{
log: m.log,
topic: topic,
scale: float64(scale),
timeout: timeout,
log: m.log,
mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }),
topic: topic,
scale: float64(scale),
}
m.Listen(topic, h.Receive)
@ -133,9 +130,9 @@ func (m *MqttClient) IntGetter(topic string, scale int64, timeout time.Duration)
// StringGetter creates handler for string from MQTT topic that returns cached value
func (m *MqttClient) StringGetter(topic string, timeout time.Duration) StringGetter {
h := &msgHandler{
log: m.log,
topic: topic,
timeout: timeout,
log: m.log,
mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }),
topic: topic,
}
m.Listen(topic, h.Receive)
@ -145,9 +142,9 @@ func (m *MqttClient) StringGetter(topic string, timeout time.Duration) StringGet
// BoolGetter creates handler for string from MQTT topic that returns cached value
func (m *MqttClient) BoolGetter(topic string, timeout time.Duration) BoolGetter {
h := &msgHandler{
log: m.log,
topic: topic,
timeout: timeout,
log: m.log,
mux: util.NewWaiter(timeout, func() { m.log.TRACE.Printf("%s wait for initial value", topic) }),
topic: topic,
}
m.Listen(topic, h.Receive)
@ -214,10 +211,7 @@ func (m *MqttClient) WaitForToken(token mqtt.Token) {
type msgHandler struct {
log *util.Logger
once sync.Once
mux sync.Mutex
updated time.Time
timeout time.Duration
mux *util.Waiter
scale float64
topic string
payload string
@ -230,31 +224,14 @@ func (h *msgHandler) Receive(payload string) {
defer h.mux.Unlock()
h.payload = payload
h.updated = time.Now()
}
func (h *msgHandler) waitForInitialValue() {
h.mux.Lock()
defer h.mux.Unlock()
if h.updated.IsZero() {
h.log.TRACE.Printf("%s wait for initial value", h.topic)
// wait for initial update
for h.updated.IsZero() {
h.mux.Unlock()
time.Sleep(waitTimeout)
h.mux.Lock()
}
}
h.mux.Update()
}
func (h *msgHandler) hasValue() (string, error) {
h.once.Do(h.waitForInitialValue)
h.mux.Lock()
elapsed := h.mux.LockWithTimeout()
defer h.mux.Unlock()
if elapsed := time.Since(h.updated); h.timeout != 0 && elapsed > h.timeout {
if elapsed > 0 {
return "", fmt.Errorf("%s outdated: %v", h.topic, elapsed.Truncate(time.Second))
}

33
provider/socket.go
View file

@ -6,7 +6,6 @@ import (
"math"
"net/http"
"strconv"
"sync"
"time"
"github.com/andig/evcc/util"
@ -18,15 +17,12 @@ import (
// Socket implements websocket request provider
type Socket struct {
*util.HTTPHelper
mux sync.Mutex
once sync.Once
mux *util.Waiter
url string
headers map[string]string
scale float64
jq *gojq.Query
timeout time.Duration
val interface{}
updated time.Time
}
// NewSocketProviderFromConfig creates a HTTP provider
@ -46,10 +42,10 @@ func NewSocketProviderFromConfig(log *util.Logger, other map[string]interface{})
p := &Socket{
HTTPHelper: util.NewHTTPHelper(logger),
mux: util.NewWaiter(cc.Timeout, func() { logger.TRACE.Println("wait for initial value") }),
url: cc.URI,
headers: cc.Headers,
scale: cc.Scale,
timeout: cc.Timeout,
}
// handle basic auth
@ -110,41 +106,24 @@ func (p *Socket) listen() {
v, err := jq.Query(p.jq, b)
if err == nil {
p.val = v
p.updated = time.Now()
p.mux.Update()
} else {
log.WARN.Printf("invalid: %s", string(b))
}
} else {
p.val = string(b)
p.updated = time.Now()
p.mux.Update()
}
p.mux.Unlock()
}
}
}
func (p *Socket) waitForInitialValue() {
p.mux.Lock()
defer p.mux.Unlock()
if p.updated.IsZero() {
p.HTTPHelper.Log.TRACE.Println("wait for initial value")
// wait for initial update
for p.updated.IsZero() {
p.mux.Unlock()
time.Sleep(waitTimeout)
p.mux.Lock()
}
}
}
func (p *Socket) hasValue() (interface{}, error) {
p.once.Do(p.waitForInitialValue)
p.mux.Lock()
elapsed := p.mux.LockWithTimeout()
defer p.mux.Unlock()
if elapsed := time.Since(p.updated); p.timeout != 0 && elapsed > p.timeout {
if elapsed > 0 {
return nil, fmt.Errorf("outdated: %v", elapsed.Truncate(time.Second))
}

60
util/waiter.go Normal file
View file

@ -0,0 +1,60 @@
package util
import (
"sync"
"time"
)
const waitTimeout = 50 * time.Millisecond // polling interval when waiting for initial value
// Waiter provides monitoring of receive timeouts and reception of initial value
type Waiter struct {
sync.Mutex
log func()
once sync.Once
updated time.Time
timeout time.Duration
}
// NewWaiter creates new waiter
func NewWaiter(timeout time.Duration, logInitialWait func()) *Waiter {
return &Waiter{
log: logInitialWait,
timeout: timeout,
}
}
// Update is called when client has received data. Update resets the timeout counter.
// It is client responsibility to ensure that the waiter is not locked when Update is called.
func (p *Waiter) Update() {
p.updated = time.Now()
}
// waitForInitialValue blocks until Update has been called at least once.
// It assumes lock has been obtained before and returns with lock active.
func (p *Waiter) waitForInitialValue() {
if p.updated.IsZero() {
p.log()
// wait for initial update
for p.updated.IsZero() {
p.Unlock()
time.Sleep(waitTimeout)
p.Lock()
}
}
}
// LockWithTimeout waits for initial value and checks if update timeout has elapsed
func (p *Waiter) LockWithTimeout() time.Duration {
p.Lock()
// waiting assumes lock acquired and returns with lock
p.once.Do(p.waitForInitialValue)
if elapsed := time.Since(p.updated); p.timeout != 0 && elapsed > p.timeout {
return elapsed
}
return 0
}