[feat] mqtt

cmd/homectrl/main.go

@@ -30,5 +30,6 @@ func main() {
 	}()
 	var inst = zigbee2mqtt.Zigbee2mqtt{}
 	inst.New(broker, port)
+	fmt.Println(inst.Devices)
 
 }

internal/connectors/mqtt/mqtt.go

@@ -1,7 +1,6 @@
 package mqtt
 
 import (
-	"flag"
 	"fmt"
 
 	"container/list"
@@ -39,14 +38,17 @@ var ConnectLostHandler mqtt_cli.ConnectionLostHandler = func(client mqtt_cli.Cli
 	fmt.Printf("Connect lost: %v", err)
 }
 
-func (mq MqttInst) subscribeTopic(topic string) {
-	token := mq.client.Subscribe(topic, 1, nil)
+func (mq MqttInst) subscribeTopic(topic string, callback mqtt_cli.MessageHandler) {
+	token := mq.client.Subscribe(topic, 1, callback)
 	token.Wait()
-	fmt.Printf("Subscribed to topic: %s", topic)
+	fmt.Printf("Subscribed to topic: %s\n", topic)
 	receiveCount := 0
 	choke := make(chan [2]string)
-	num := flag.Int("num", 1, "The number of messages to publish or subscribe (default 1)")
-	for receiveCount < *num {
+	var num = 1
+	for receiveCount < num {
+
+		receiveCount++
+
 		incoming := <-choke
 		fmt.Printf("RECEIVED TOPIC: %s MESSAGE: %s\n", incoming[0], incoming[1])
 		//devices.Builder2(incoming[1])
@@ -54,11 +56,12 @@ func (mq MqttInst) subscribeTopic(topic string) {
 	}
 }
 
-func (mq *MqttInst) setDefaultOptions() {
-	mq.options.AddBroker(fmt.Sprintf("tcp://%s:%d", mq.broker, mq.port))
+func (mq *MqttInst) SetDefaultOptions() {
+	mq.options.SetDefaultPublishHandler(messagePubHandler)
+	mq.options.AddBroker(fmt.Sprintf("mqtt://%s:%d", mq.broker, mq.port))
 	mq.options.SetClientID("go_mqtt_client")
-	mq.options.SetUsername("emqx")
-	mq.options.SetPassword("public")
+	/* mq.options.SetUsername("emqx")
+	mq.options.SetPassword("public") */
 	mq.options.SetDefaultPublishHandler(messagePubHandler)
 	mq.options.OnConnect = ConnectHandler
 	mq.options.OnConnectionLost = ConnectLostHandler
@@ -66,10 +69,10 @@ func (mq *MqttInst) setDefaultOptions() {
 
 func (mq *MqttInst) SetDefaultPublishHandler(messagePubHandler mqtt_cli.MessageHandler) {
 	mq.options.SetDefaultPublishHandler(messagePubHandler)
-	mq.options.AddBroker(fmt.Sprintf("tcp://%s:%d", mq.broker, mq.port))
-	mq.options.SetClientID("go_mqtt_client")
-	mq.options.SetUsername("emqx")
-	mq.options.SetPassword("public")
+	mq.options.AddBroker(fmt.Sprintf("mqtt://%s:%d", mq.broker, mq.port))
+	mq.options.SetClientID("HomeCtrl")
+	/* mq.options.SetUsername("emqx")
+	mq.options.SetPassword("public") */
 	mq.options.OnConnect = ConnectHandler
 	mq.options.OnConnectionLost = ConnectLostHandler
 }
@@ -89,7 +92,7 @@ func (mq *MqttInst) Connect() {
 
 	mq.client = mqtt_cli.NewClient(mq.options)
 	if token := mq.client.Connect(); token.Wait() && token.Error() != nil {
-		fmt.Println("Error")
+		fmt.Println("Connection error")
 		panic(token.Error())
 	}
 }
@@ -98,15 +101,16 @@ func (mq *MqttInst) Connect() {
 func (mq *MqttInst) New(broker string, port int) {
 	fmt.Println("mqtt/New")
 	mq.options = mqtt_cli.NewClientOptions()
-	//mq.setDefaultOptions()
+
 	mq.broker = broker
 	mq.port = port
+	mq.SetDefaultOptions()
 	//mq.Connect()
 }
 
 // SubTopic subscribe to a topic
-func (mq *MqttInst) SubscribeTopic(topic string) {
+func (mq *MqttInst) SubscribeTopic(topic string, callback mqtt_cli.MessageHandler) {
 	fmt.Println("subtopic")
 	mq.topics.PushBack(topic)
-	mq.subscribeTopic(topic)
+	mq.subscribeTopic(topic, callback)
 }

internal/connectors/zigbee2mqtt/device.go

@@ -146,18 +146,11 @@ type device struct {
 	Definition          definition
 }
 
-type Builder func([]byte)
-
-func Build(data []byte) []device {
-	var arr []device
-	err := json.Unmarshal(data, &arr)
+func Build(data []byte, Z2m *Zigbee2mqtt) {
+	//var arr []device
+	err := json.Unmarshal(data, &Z2m.Devices)
 	fmt.Println("connectors/zigbee2mqtt/device/builder")
 	if err != nil {
 		panic(err)
 	}
-	fmt.Println(arr)
-	for _, s := range arr {
-		fmt.Println(s)
-	}
-	return arr
 }

internal/connectors/zigbee2mqtt/zigbee2mqtt.go

@@ -8,20 +8,25 @@ import (
 )
 
 type Zigbee2mqtt struct {
-	devices []device
+	Devices []device
 }
 
+var Z2m = &Zigbee2mqtt{}
+
 var messagePubHandler mqtt_cli.MessageHandler = func(client mqtt_cli.Client, msg mqtt_cli.Message) {
-	fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
-	Build(msg.Payload())
+	//fmt.Printf("Received message: %s from topic: %s\n", msg.Payload(), msg.Topic())
+	Build(msg.Payload(), Z2m)
 }
 
 func (z2m *Zigbee2mqtt) New(broker string, port int) {
 	fmt.Println("zigbee2mqtt/New")
 	var inst = &mqtt.MqttInst{}
 	inst.New(broker, port)
-	inst.SetDefaultPublishHandler(messagePubHandler)
+	//inst.SetDefaultPublishHandler(messagePubHandler)
 	inst.Connect()
 	fmt.Println("zigbee2mqtt/New2")
-	inst.SubscribeTopic("zigbee2mqtt/bridge/devices")
+	inst.SubscribeTopic("zigbee2mqtt/bridge/devices", messagePubHandler)
+	z2m = Z2m
+	fmt.Println("lol")
+	fmt.Println(z2m.Devices)
 }