diff --git a/provider/mqtt.go b/provider/mqtt.go index c722e15f3..a92a88b7c 100644 --- a/provider/mqtt.go +++ b/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 -} diff --git a/provider/mqtt_handler.go b/provider/mqtt_handler.go new file mode 100644 index 000000000..70144f5ca --- /dev/null +++ b/provider/mqtt_handler.go @@ -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 +}