From 0490db10fa23c520367488f22e279a3fc4bcb380 Mon Sep 17 00:00:00 2001 From: andig Date: Wed, 20 May 2020 15:54:16 +0200 Subject: [PATCH] Add util.Waiter for initial value and timeout handling (#174) --- meter/sma.go | 47 +++++++++-------------------------- meter/sma_test.go | 1 + provider/mqtt.go | 61 +++++++++++++++------------------------------- provider/socket.go | 33 +++++-------------------- util/waiter.go | 60 +++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 98 insertions(+), 104 deletions(-) create mode 100644 util/waiter.go diff --git a/meter/sma.go b/meter/sma.go index 08ab2180d..8ca93cad1 100644 --- a/meter/sma.go +++ b/meter/sma.go @@ -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)) } diff --git a/meter/sma_test.go b/meter/sma_test.go index 901288459..30595c16d 100644 --- a/meter/sma_test.go +++ b/meter/sma_test.go @@ -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) diff --git a/provider/mqtt.go b/provider/mqtt.go index 2477a8aa6..952be7a19 100644 --- a/provider/mqtt.go +++ b/provider/mqtt.go @@ -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)) } diff --git a/provider/socket.go b/provider/socket.go index caa132be0..500628021 100644 --- a/provider/socket.go +++ b/provider/socket.go @@ -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)) } diff --git a/util/waiter.go b/util/waiter.go new file mode 100644 index 000000000..94e358ddf --- /dev/null +++ b/util/waiter.go @@ -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 +}