前端订阅

broker/hook.go

@@ -38,39 +38,52 @@ func (h *Hook) OnACLCheck(cl *mqtt.Client, topic string, write bool) bool {
 }
 
 func (h *Hook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
-
+	//执行unsubscribe
+	subs := cl.State.Subscriptions.GetAll()
+	for _, sub := range subs {
+		handleUnsubscribe(sub.Filter)
+	}
 }
 
 func (h *Hook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte) {
 	//device/+/values
 	//project/+/values
+	//space/+/values
 	for _, f := range pk.Filters {
-		ss := strings.Split(f.Filter, "/")
-		if len(ss) == 3 {
-			switch ss[0] {
-			case "device":
-				watchDeviceValues(ss[1])
-			case "project":
-				watchProjectValues(ss[1])
-			case "space":
-				watchSpaceValues(ss[1])
-			}
-		}
+		handleSubscribe(f.Filter)
 	}
 }
 
 func (h *Hook) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet) {
 	for _, f := range pk.Filters {
-		ss := strings.Split(f.Filter, "/")
-		if len(ss) == 3 {
-			switch ss[0] {
-			case "device":
-				unWatchDeviceValues(ss[1])
-			case "project":
-				unWatchProjectValues(ss[1])
-			case "space":
-				unWatchSpaceValues(ss[1])
-			}
+		handleUnsubscribe(f.Filter)
+	}
+}
+
+func handleSubscribe(filter string) {
+	ss := strings.Split(filter, "/")
+	if len(ss) == 3 && ss[2] == "values" {
+		switch ss[0] {
+		case "device":
+			watchDeviceValues(ss[1])
+		case "project":
+			watchProjectValues(ss[1])
+		case "space":
+			watchSpaceValues(ss[1])
+		}
+	}
+}
+
+func handleUnsubscribe(filter string) {
+	ss := strings.Split(filter, "/")
+	if len(ss) == 3 && ss[2] == "values" {
+		switch ss[0] {
+		case "device":
+			unWatchDeviceValues(ss[1])
+		case "project":
+			unWatchProjectValues(ss[1])
+		case "space":
+			unWatchSpaceValues(ss[1])
 		}
 	}
 }

types/options.go

@@ -0,0 +1,136 @@
+package types
+
+import "strconv"
+
+type Options map[string]any
+
+func (o Options) Float64(name string, def float64) float64 {
+	if value, ok := o[name]; ok {
+		switch value.(type) {
+		case bool:
+			if value.(bool) {
+				def = 1
+			} else {
+				def = 0
+			}
+		case uint8:
+			def = float64(value.(uint8))
+		case uint16:
+			def = float64(value.(uint16))
+		case uint32:
+			def = float64(value.(uint32))
+		case uint64:
+			def = float64(value.(uint64))
+		case uint:
+			def = float64(value.(uint))
+		case int8:
+			def = float64(value.(int8))
+		case int16:
+			def = float64(value.(int16))
+		case int32:
+			def = float64(value.(int32))
+		case int64:
+			def = float64(value.(int64))
+		case int:
+			def = float64(value.(int))
+		case float32:
+			def = float64(value.(float32))
+		case float64:
+			def = value.(float64)
+		case string:
+			v, e := strconv.ParseFloat(value.(string), 64)
+			if e == nil {
+				def = v
+			}
+		}
+	}
+	return def
+}
+
+func (o Options) Int64(name string, def int64) int64 {
+	if value, ok := o[name]; ok {
+		switch value.(type) {
+		case bool:
+			if value.(bool) {
+				def = 1
+			} else {
+				def = 0
+			}
+		case uint8:
+			def = int64(value.(uint8))
+		case uint16:
+			def = int64(value.(uint16))
+		case uint32:
+			def = int64(value.(uint32))
+		case uint64:
+			def = int64(value.(uint64))
+		case uint:
+			def = int64(value.(uint))
+		case int8:
+			def = int64(value.(int8))
+		case int16:
+			def = int64(value.(int16))
+		case int32:
+			def = int64(value.(int32))
+		case int64:
+			def = value.(int64)
+		case int:
+			def = int64(value.(int))
+		case float32:
+			def = int64(value.(float32))
+		case float64:
+			def = int64(value.(float64))
+		case string:
+			v, e := strconv.ParseInt(value.(string), 10, 64)
+			if e == nil {
+				def = v
+			}
+		}
+	}
+	return def
+}
+
+func (o Options) Int(name string, def int) int {
+	return int(o.Int64(name, int64(def)))
+}
+
+func (o Options) Bool(name string, def bool) bool {
+	if value, ok := o[name]; ok {
+		switch value.(type) {
+		case bool:
+			def = value.(bool)
+		case uint8:
+			def = value.(uint8) != 0
+		case uint16:
+			def = value.(uint16) != 0
+		case uint32:
+			def = value.(uint32) != 0
+		case uint64:
+			def = value.(uint64) != 0
+		case uint:
+			def = value.(uint) != 0
+		case int8:
+			def = value.(int8) != 0
+		case int16:
+			def = value.(int16) != 0
+		case int32:
+			def = value.(int32) != 0
+		case int64:
+			def = value.(int64) != 0
+		case int:
+			def = value.(int) != 0
+		case float32:
+			def = value.(float32) != 0
+		case float64:
+			def = value.(float64) != 0
+		case string:
+			switch value.(string) {
+			case "true", "TRUE", "True", "1":
+				def = true
+			default:
+				def = false
+			}
+		}
+	}
+	return def
+}