louis/oktopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(device status): change topic communication to allow retained messages

Browse Source
This commit is contained in:
Leandro Antônio Farias Machado 2023-07-02 15:57:13 -03:00
parent 5f64795225
commit 9924ad50c6
3 changed files with 16 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
backend/services/controller/cmd/oktopus/main.go
View File

@ -45,8 +45,8 @@ func main() {
log.Println("Starting Oktopus Project TR-369 Controller Version:", VERSION) log.Println("Starting Oktopus Project TR-369 Controller Version:", VERSION)
// fl_endpointId := flag.String("endpoint_id", "proto::oktopus-controller", "Defines the enpoint id the Agent must trust on.") // fl_endpointId := flag.String("endpoint_id", "proto::oktopus-controller", "Defines the enpoint id the Agent must trust on.")
flDevicesTopic := flag.String("d", "oktopus/devices", "That's the topic mqtt broker end new devices info.") flDevicesTopic := flag.String("d", "oktopus/+/devices/+", "That's the topic mqtt broker end new devices info.")
flDisconTopic := flag.String("dis", "oktopus/disconnect", "It's where disconnected IoTs are known.") flDisconTopic := flag.String("dis", "oktopus/+/disconnect/+", "It's where disconnected IoTs are known.")
flSubTopic := flag.String("sub", "oktopus/+/controller/+", "That's the topic agent must publish to, and the controller keeps on listening.") flSubTopic := flag.String("sub", "oktopus/+/controller/+", "That's the topic agent must publish to, and the controller keeps on listening.")
flBrokerAddr := flag.String("a", "localhost", "Mqtt broker adrress") flBrokerAddr := flag.String("a", "localhost", "Mqtt broker adrress")
flBrokerPort := flag.String("p", "1883", "Mqtt broker port") flBrokerPort := flag.String("p", "1883", "Mqtt broker port")

18
backend/services/controller/internal/mqtt/mqtt.go
View File

@ -126,11 +126,11 @@ func (m *Mqtt) Publish(msg []byte, topic, respTopic string) {
func (m *Mqtt) buildClientConfig(devices, controller, disconnect, apiMsg chan *paho.Publish) *paho.ClientConfig { func (m *Mqtt) buildClientConfig(devices, controller, disconnect, apiMsg chan *paho.Publish) *paho.ClientConfig {
log.Println("Starting new mqtt client") log.Println("Starting new mqtt client")
singleHandler := paho.NewSingleHandlerRouter(func(p *paho.Publish) { singleHandler := paho.NewSingleHandlerRouter(func(p *paho.Publish) {
if p.Topic == m.DevicesTopic { if strings.Contains(p.Topic, "devices") {
devices <- p devices <- p
} else if strings.Contains(p.Topic, "controller") { } else if strings.Contains(p.Topic, "controller") {
controller <- p controller <- p
} else if p.Topic == m.DisconnectTopic { } else if strings.Contains(p.Topic, "disconnect") {
disconnect <- p disconnect <- p
} else if strings.Contains(p.Topic, "api") { } else if strings.Contains(p.Topic, "api") {
apiMsg <- p apiMsg <- p
@ -173,17 +173,19 @@ func (m *Mqtt) messageHandler(devices, controller, disconnect, apiMsg chan *paho
for { for {
select { select {
case d := <-devices: case d := <-devices:
payload := string(d.Payload) paths := strings.Split(d.Topic, "/")
log.Println("New device: ", payload) device := paths[len(paths)-1]
m.handleNewDevice(payload) log.Println("New device: ", device)
m.handleNewDevice(device)
case c := <-controller: case c := <-controller:
topic := c.Topic topic := c.Topic
sn := strings.Split(topic, "/") sn := strings.Split(topic, "/")
m.handleNewDevicesResponse(c.Payload, sn[3]) m.handleNewDevicesResponse(c.Payload, sn[3])
case dis := <-disconnect: case dis := <-disconnect:
payload := string(dis.Payload) paths := strings.Split(dis.Topic, "/")
log.Println("Device disconnected: ", payload) device := paths[len(paths)-1]
m.handleDevicesDisconnect(payload) log.Println("Device disconnected: ", device)
m.handleDevicesDisconnect(device)
case api := <-apiMsg: case api := <-apiMsg:
log.Println("Handle api request") log.Println("Handle api request")
m.handleApiRequest(api.Payload) m.handleApiRequest(api.Payload)

8
backend/services/mochi/cmd/main.go
View File

@ -252,7 +252,7 @@ func (h *MyHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
} }
if clUser != "" { if clUser != "" {
err := server.Publish("oktopus/disconnect", []byte(clUser), false, 1) err := server.Publish("oktopus/v1/disconnect/"+clUser, []byte(""), false, 1)
if err != nil { if err != nil {
log.Println("server publish error: ", err) log.Println("server publish error: ", err)
} }
@ -270,11 +270,11 @@ func (h *MyHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []
if clUser != "" { if clUser != "" {
cl.Properties.Will = mqtt.Will{ cl.Properties.Will = mqtt.Will{
Qos: 1, Qos: 1,
Payload: []byte(clUser), TopicName: "oktopus/v1/disconnect/" + clUser,
} }
log.Println("new device:", clUser) log.Println("new device:", clUser)
err := server.Publish("oktopus/devices", []byte(clUser), false, 1) err := server.Publish("oktopus/v1/devices/"+clUser, []byte(""), false, 1)
if err != nil { if err != nil {
log.Println("server publish error: ", err) log.Println("server publish error: ", err)
} }