异步处理命令响应

This commit is contained in:
ctlove0523 2021-03-01 23:51:07 +08:00
parent 9d7aa65df5
commit fd6ffc19b9
2 changed files with 22 additions and 21 deletions

View File

@ -366,28 +366,28 @@ func (device *iotDevice) createMessageMqttHandler() func(client mqtt.Client, mes
func (device *iotDevice) createCommandMqttHandler() func(client mqtt.Client, message mqtt.Message) { func (device *iotDevice) createCommandMqttHandler() func(client mqtt.Client, message mqtt.Message) {
commandHandler := func(client mqtt.Client, message mqtt.Message) { commandHandler := func(client mqtt.Client, message mqtt.Message) {
command := &Command{}
if json.Unmarshal(message.Payload(), command) != nil {
glog.Warningf("unmarshal platform command failed,device id = %smessage = %s", device.Id, message)
}
handleFlag := true
for _, handler := range device.commandHandlers {
handleFlag = handleFlag && handler(*command)
}
var res string
if handleFlag {
glog.Infof("device %s handle command success", device.Id)
res = Interface2JsonString(CommandResponse{
ResultCode: 0,
})
} else {
glog.Warningf("device %s handle command failed", device.Id)
res = Interface2JsonString(CommandResponse{
ResultCode: 1,
})
}
go func() { go func() {
command := &Command{}
if json.Unmarshal(message.Payload(), command) != nil {
glog.Warningf("unmarshal platform command failed,device id = %smessage = %s", device.Id, message)
}
handleFlag := true
for _, handler := range device.commandHandlers {
handleFlag = handleFlag && handler(*command)
}
var res string
if handleFlag {
glog.Infof("device %s handle command success", device.Id)
res = Interface2JsonString(CommandResponse{
ResultCode: 0,
})
} else {
glog.Warningf("device %s handle command failed", device.Id)
res = Interface2JsonString(CommandResponse{
ResultCode: 1,
})
}
if token := device.client.Publish(FormatTopic(CommandResponseTopic, device.Id)+GetTopicRequestId(message.Topic()), 1, false, res); if token := device.client.Publish(FormatTopic(CommandResponseTopic, device.Id)+GetTopicRequestId(message.Topic()), 1, false, res);
token.Wait() && token.Error() != nil { token.Wait() && token.Error() != nil {
glog.Infof("device %s send command response failed", device.Id) glog.Infof("device %s send command response failed", device.Id)

View File

@ -22,6 +22,7 @@ func main() {
// 添加用于处理平台下发命令的callback // 添加用于处理平台下发命令的callback
device.AddCommandHandler(func(command iot.Command) bool { device.AddCommandHandler(func(command iot.Command) bool {
fmt.Println("I get command from platform") fmt.Println("I get command from platform")
time.Sleep(10 * time.Second)
return true return true
}) })
time.Sleep(10 * time.Minute) time.Sleep(10 * time.Minute)