observable connection state

base.go

@@ -46,7 +46,7 @@ type stateField[Type any] struct {
 }
 
 func (state *stateField[Type]) send(command string) {
-	state.actuator.send(fmt.Sprintf("{\"%s\":%s}", state.key, command))
+	state.actuator.send(fmt.Sprintf("{\"%s\":\"%s\"}", state.key, command))
 }
 
 func (d *stateField[_]) Close() error {
@@ -91,7 +91,7 @@ func (field *onOffStateField) Process(state map[string]interface{}) {
 			field.values <- true
 		}
 		if strings.EqualFold("OFF", str) {
-			field.values <- true
+			field.values <- false
 		}
 	}
 }

paho/pubsub.go

@@ -1,22 +1,32 @@
 package paho
 
 import (
+	"log"
+
 	"github.com/balazsgrill/hass"
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 )
 
 type PahoPubSub struct {
 	mqtt.Client
 	subscriptions map[string][]func(topic string, payload []byte)
+	connstate     chan bool
 }
 
 var _ hass.IPubSubConnecterRuntime = &PahoPubSub{}
 
 func New(options *mqtt.ClientOptions) hass.IPubSubConnecterRuntime {
 	result := &PahoPubSub{
 		subscriptions: make(map[string][]func(topic string, payload []byte)),
+		connstate:     make(chan bool),
 	}
-	options = options.SetOnConnectHandler(result.onConnect)
+	options = options.SetOnConnectHandler(result.onConnect).SetConnectionLostHandler(func(c mqtt.Client, err error) {
+		if err != nil {
+			log.Println(err)
+		}
+		result.connstate <- false
+	}).SetAutoReconnect(true)
+
 	result.Client = mqtt.NewClient(options)
 	return result
 }
@@ -27,6 +37,11 @@ func (ps *PahoPubSub) onConnect(client mqtt.Client) {
 			subscribe(ps.Client, topic, callback)
 		}
 	}
+	ps.connstate <- true
+}
+
+func (ps *PahoPubSub) ConnectionState() chan bool {
+	return ps.connstate
 }
 
 func (ps *PahoPubSub) Send(topic string, payload []byte) error {

pubsub.go

@@ -8,6 +8,7 @@ type IPubSubRuntime interface {
 type IConnecter interface {
 	Connect() error
 	Disconnect()
+	ConnectionState() chan bool
 }
 
 type IPubSubConnecterRuntime interface {