Mqtt: fix handler not returning when initial value not received (#2008)
This commit is contained in:
andig
GitHub
parent
6dcd0aa976
commit
94e2a28839
2 changed files with 116 additions and 113 deletions
133
provider/mqtt.go
133
provider/mqtt.go
|
|
@ -2,14 +2,12 @@ package provider
|
|||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/evcc-io/evcc/provider/mqtt"
|
||||
"github.com/evcc-io/evcc/util"
|
||||
"github.com/evcc-io/evcc/util/jq"
|
||||
"github.com/evcc-io/evcc/util/request"
|
||||
"github.com/itchyny/gojq"
|
||||
)
|
||||
|
||||
|
|
@ -115,17 +113,30 @@ func (m *Mqtt) WithJq(jq string) (*Mqtt, error) {
|
|||
|
||||
var _ FloatProvider = (*Mqtt)(nil)
|
||||
|
||||
// FloatGetter creates handler for float64 from MQTT topic that returns cached value
|
||||
func (m *Mqtt) FloatGetter() func() (float64, error) {
|
||||
// newReceiver creates a msgHandler and subscribes it to the topic.
|
||||
// receiver will ensure actual data guarded by `timeout` and return error
|
||||
// if initial value is not received within `timeout` or max. 10s if timeout is not given.
|
||||
func (m *Mqtt) newReceiver() *msgHandler {
|
||||
wait := m.timeout
|
||||
if wait == 0 {
|
||||
wait = request.Timeout
|
||||
}
|
||||
|
||||
h := &msgHandler{
|
||||
topic: m.topic,
|
||||
scale: m.scale,
|
||||
mux: util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
|
||||
mux: util.NewWaiter(wait, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
|
||||
re: m.re,
|
||||
jq: m.jq,
|
||||
}
|
||||
|
||||
m.client.Listen(m.topic, h.receive)
|
||||
return h
|
||||
}
|
||||
|
||||
// FloatGetter creates handler for float64 from MQTT topic that returns cached value
|
||||
func (m *Mqtt) FloatGetter() func() (float64, error) {
|
||||
h := m.newReceiver()
|
||||
return h.floatGetter
|
||||
}
|
||||
|
||||
|
|
@ -133,15 +144,7 @@ var _ IntProvider = (*Mqtt)(nil)
|
|||
|
||||
// IntGetter creates handler for int64 from MQTT topic that returns cached value
|
||||
func (m *Mqtt) IntGetter() func() (int64, error) {
|
||||
h := &msgHandler{
|
||||
topic: m.topic,
|
||||
scale: m.scale,
|
||||
mux: util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
|
||||
re: m.re,
|
||||
jq: m.jq,
|
||||
}
|
||||
|
||||
m.client.Listen(m.topic, h.receive)
|
||||
h := m.newReceiver()
|
||||
return h.intGetter
|
||||
}
|
||||
|
||||
|
|
@ -149,14 +152,7 @@ var _ StringProvider = (*Mqtt)(nil)
|
|||
|
||||
// StringGetter creates handler for string from MQTT topic that returns cached value
|
||||
func (m *Mqtt) StringGetter() func() (string, error) {
|
||||
h := &msgHandler{
|
||||
topic: m.topic,
|
||||
mux: util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
|
||||
re: m.re,
|
||||
jq: m.jq,
|
||||
}
|
||||
|
||||
m.client.Listen(m.topic, h.receive)
|
||||
h := m.newReceiver()
|
||||
return h.stringGetter
|
||||
}
|
||||
|
||||
|
|
@ -164,14 +160,7 @@ var _ BoolProvider = (*Mqtt)(nil)
|
|||
|
||||
// BoolGetter creates handler for string from MQTT topic that returns cached value
|
||||
func (m *Mqtt) BoolGetter() func() (bool, error) {
|
||||
h := &msgHandler{
|
||||
topic: m.topic,
|
||||
mux: util.NewWaiter(m.timeout, func() { m.log.DEBUG.Printf("%s wait for initial value", m.topic) }),
|
||||
re: m.re,
|
||||
jq: m.jq,
|
||||
}
|
||||
|
||||
m.client.Listen(m.topic, h.receive)
|
||||
h := m.newReceiver()
|
||||
return h.boolGetter
|
||||
}
|
||||
|
||||
|
|
@ -216,85 +205,3 @@ func (m *Mqtt) StringSetter(param string) func(string) error {
|
|||
return m.client.Publish(m.topic, false, payload)
|
||||
}
|
||||
}
|
||||
|
||||
type msgHandler struct {
|
||||
mux *util.Waiter
|
||||
scale float64
|
||||
topic string
|
||||
payload string
|
||||
re *regexp.Regexp
|
||||
jq *gojq.Query
|
||||
}
|
||||
|
||||
func (h *msgHandler) receive(payload string) {
|
||||
h.mux.Lock()
|
||||
defer h.mux.Unlock()
|
||||
|
||||
h.payload = payload
|
||||
h.mux.Update()
|
||||
}
|
||||
|
||||
func (h *msgHandler) hasValue() (string, error) {
|
||||
elapsed := h.mux.LockWithTimeout()
|
||||
defer h.mux.Unlock()
|
||||
|
||||
if elapsed > 0 {
|
||||
return "", fmt.Errorf("%s outdated: %v", h.topic, elapsed.Truncate(time.Second))
|
||||
}
|
||||
|
||||
var err error
|
||||
payload := h.payload
|
||||
|
||||
if h.re != nil {
|
||||
m := h.re.FindStringSubmatch(payload)
|
||||
if len(m) > 1 {
|
||||
payload = m[1] // first submatch
|
||||
}
|
||||
}
|
||||
|
||||
if h.jq != nil {
|
||||
var val interface{}
|
||||
if val, err = jq.Query(h.jq, []byte(payload)); err == nil {
|
||||
payload = fmt.Sprintf("%v", val)
|
||||
}
|
||||
}
|
||||
|
||||
return payload, err
|
||||
}
|
||||
|
||||
func (h *msgHandler) floatGetter() (float64, error) {
|
||||
v, err := h.hasValue()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
f, err := strconv.ParseFloat(v, 64)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s invalid: '%s'", h.topic, v)
|
||||
}
|
||||
|
||||
return f * h.scale, nil
|
||||
}
|
||||
|
||||
func (h *msgHandler) intGetter() (int64, error) {
|
||||
f, err := h.floatGetter()
|
||||
return int64(math.Round(f)), err
|
||||
}
|
||||
|
||||
func (h *msgHandler) stringGetter() (string, error) {
|
||||
v, err := h.hasValue()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return string(v), nil
|
||||
}
|
||||
|
||||
func (h *msgHandler) boolGetter() (bool, error) {
|
||||
v, err := h.hasValue()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
return util.Truish(v), nil
|
||||
}
|
||||
|
|
|
|||
96
provider/mqtt_handler.go
Normal file
96
provider/mqtt_handler.go
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
package provider
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/evcc-io/evcc/util"
|
||||
"github.com/evcc-io/evcc/util/jq"
|
||||
"github.com/itchyny/gojq"
|
||||
)
|
||||
|
||||
type msgHandler struct {
|
||||
mux *util.Waiter
|
||||
scale float64
|
||||
topic string
|
||||
payload string
|
||||
re *regexp.Regexp
|
||||
jq *gojq.Query
|
||||
}
|
||||
|
||||
func (h *msgHandler) receive(payload string) {
|
||||
h.mux.Lock()
|
||||
defer h.mux.Unlock()
|
||||
|
||||
h.payload = payload
|
||||
h.mux.Update()
|
||||
}
|
||||
|
||||
// hasValue returned the received and processed payload as string
|
||||
func (h *msgHandler) hasValue() (string, error) {
|
||||
elapsed := h.mux.LockWithTimeout()
|
||||
defer h.mux.Unlock()
|
||||
|
||||
if elapsed > 0 {
|
||||
return "", fmt.Errorf("%s outdated: %v", h.topic, elapsed.Truncate(time.Second))
|
||||
}
|
||||
|
||||
var err error
|
||||
payload := h.payload
|
||||
|
||||
if h.re != nil {
|
||||
m := h.re.FindStringSubmatch(payload)
|
||||
if len(m) > 1 {
|
||||
payload = m[1] // first submatch
|
||||
}
|
||||
}
|
||||
|
||||
if h.jq != nil {
|
||||
var val interface{}
|
||||
if val, err = jq.Query(h.jq, []byte(payload)); err == nil {
|
||||
payload = fmt.Sprintf("%v", val)
|
||||
}
|
||||
}
|
||||
|
||||
return payload, err
|
||||
}
|
||||
|
||||
func (h *msgHandler) floatGetter() (float64, error) {
|
||||
v, err := h.hasValue()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
f, err := strconv.ParseFloat(v, 64)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s invalid: '%s'", h.topic, v)
|
||||
}
|
||||
|
||||
return f * h.scale, nil
|
||||
}
|
||||
|
||||
func (h *msgHandler) intGetter() (int64, error) {
|
||||
f, err := h.floatGetter()
|
||||
return int64(math.Round(f)), err
|
||||
}
|
||||
|
||||
func (h *msgHandler) stringGetter() (string, error) {
|
||||
v, err := h.hasValue()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
return string(v), nil
|
||||
}
|
||||
|
||||
func (h *msgHandler) boolGetter() (bool, error) {
|
||||
v, err := h.hasValue()
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
return util.Truish(v), nil
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue