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

Fix mqtt logging and return values (#739)

Browse source
This commit is contained in:
andig 2021-03-09 14:58:43 +01:00 • committed by GitHub
parent 5018ce4b79
commit d21d15e9d7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 33 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

10
api/error.go
View file

@ -4,3 +4,13 @@ import "errors"
// ErrNotAvailable indicates that a feature is not available
var ErrNotAvailable = errors.New("not available")
// ErrTimeout is the error returned when a timeout happened.
// Modeled after context.DeadlineError
var ErrTimeout error = errTimeoutError{}
type errTimeoutError struct{}
func (errTimeoutError) Error() string { return "timeout" }
func (errTimeoutError) Timeout() bool { return true }
func (errTimeoutError) Temporary() bool { return true }

21
provider/mqtt.go
View file

@ -67,7 +67,6 @@ func NewMqtt(log *util.Logger, client *mqtt.Client, topic string, payload string
// FloatGetter creates handler for float64 from MQTT topic that returns cached value
func (m *Mqtt) FloatGetter() func() (float64, error) {
h := &msgHandler{
log: m.log,
topic: m.topic,
scale: m.scale,
mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }),
@ -80,7 +79,6 @@ func (m *Mqtt) FloatGetter() func() (float64, error) {
// IntGetter creates handler for int64 from MQTT topic that returns cached value
func (m *Mqtt) IntGetter() func() (int64, error) {
h := &msgHandler{
log: m.log,
topic: m.topic,
scale: float64(m.scale),
mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }),
@ -93,7 +91,6 @@ func (m *Mqtt) IntGetter() func() (int64, error) {
// StringGetter creates handler for string from MQTT topic that returns cached value
func (m *Mqtt) StringGetter() func() (string, error) {
h := &msgHandler{
log: m.log,
topic: m.topic,
mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }),
}
@ -105,7 +102,6 @@ func (m *Mqtt) StringGetter() func() (string, error) {
// BoolGetter creates handler for string from MQTT topic that returns cached value
func (m *Mqtt) BoolGetter() func() (bool, error) {
h := &msgHandler{
log: m.log,
topic: m.topic,
mux: util.NewWaiter(m.timeout, func() { m.log.TRACE.Printf("%s wait for initial value", m.topic) }),
}
@ -122,7 +118,6 @@ func (m *Mqtt) IntSetter(param string) func(int64) error {
return err
}
m.log.TRACE.Printf("send %s: '%s'", m.topic, payload)
return m.client.Publish(m.topic, false, payload)
}
}
@ -135,13 +130,23 @@ func (m *Mqtt) BoolSetter(param string) func(bool) error {
return err
}
m.log.TRACE.Printf("send %s: '%s'", m.topic, payload)
return m.client.Publish(m.topic, false, payload)
}
}
// StringSetter invokes script with parameter replaced by string value
func (m *Mqtt) StringSetter(param string) func(string) error {
return func(v string) error {
payload, err := setFormattedValue(m.payload, param, v)
if err != nil {
return err
}
return m.client.Publish(m.topic, false, payload)
}
}
type msgHandler struct {
log *util.Logger
mux *util.Waiter
scale float64
topic string
@ -149,8 +154,6 @@ type msgHandler struct {
}
func (h *msgHandler) receive(payload string) {
h.log.TRACE.Printf("recv %s: '%s'", h.topic, payload)
h.mux.Lock()
defer h.mux.Unlock()

13
provider/mqtt/client.go
View file

@ -6,6 +6,7 @@ import (
"sync"
"time"
"github.com/andig/evcc/api"
"github.com/andig/evcc/util"
mqtt "github.com/eclipse/paho.mqtt.golang"
)
@ -92,13 +93,14 @@ func (m *Client) ConnectionHandler(client mqtt.Client) {
}
}
// Publish synchronously pulishes payload using client qos
// Publish synchronously publishes payload using client qos
func (m *Client) Publish(topic string, retained bool, payload interface{}) error {
m.log.TRACE.Printf("send %s: '%v'", topic, payload)
token := m.Client.Publish(topic, m.Qos, retained, payload)
if token.WaitTimeout(publishTimeout) {
return token.Error()
}
return nil
return api.ErrTimeout
}
// Listen validates uniqueness and registers and attaches listener
@ -113,14 +115,15 @@ func (m *Client) Listen(topic string, callback func(string)) {
// listen attaches listener to topic
func (m *Client) listen(topic string) {
token := m.Client.Subscribe(topic, m.Qos, func(c mqtt.Client, msg mqtt.Message) {
s := string(msg.Payload())
if len(s) > 0 {
payload := string(msg.Payload())
m.log.TRACE.Printf("recv %s: '%v'", topic, payload)
if len(payload) > 0 {
m.mux.Lock()
callbacks := m.listener[topic]
m.mux.Unlock()
for _, cb := range callbacks {
cb(s)
cb(payload)
}
}
})

6
provider/mqtt/registry.go
View file

@ -47,12 +47,12 @@ func RegisteredClientOrDefault(log *util.Logger, cc Config) (*Client, error) {
var err error
client := Instance
if client == nil && cc.Broker == "" {
return nil, errors.New("missing mqtt broker configuration")
if cc.Broker != "" {
client, err = RegisteredClient(log, cc.Broker, cc.User, cc.Password, ClientID(), 1)
}
if client == nil {
client, err = RegisteredClient(log, cc.Broker, cc.User, cc.Password, ClientID(), 1)
err = errors.New("missing mqtt broker configuration")
}
return client, err