feat: support device ota upgrade

pkg/klink/klink.go

@@ -12,6 +12,7 @@ const (
 	DevLoginAction       PacketAction = "devLogin"       // 子设备上线
 	DevLogoutAction      PacketAction = "devLogout"      // 子设备下线
 	DevNetConfigAction   PacketAction = "devNetConfig"   // 设备配网信息
+	DevUpgradeAction     PacketAction = "devUpgrade"     // 设备升级
 )
 
 // DevLogin 子设备上线
@@ -66,6 +67,7 @@ type DevSend struct {
 	SubDeviceId string       `json:"subDeviceId"`
 	Data        *DevSendData `json:"data"`
 	Timestamp   int64        `json:"timestamp"`
+	Version     string       `json:"version"`
 }
 
 func (d *DevSend) Marshal() ([]byte, error) {
@@ -110,3 +112,28 @@ func (c *CloudSend) UnMarshal(bytes []byte) error {
 	}
 	return j.Struct(c)
 }
+
+// DevUpgrade 设备升级
+type DevUpgrade struct {
+	Action      string `json:"action"`
+	MsgId       int    `json:"msgId"`
+	DeviceCode  string `json:"deviceCode"`
+	SubDeviceId string `json:"subDeviceId"`
+	Timestamp   int64  `json:"timestamp"`
+	Url         string `json:"url"`
+	Md5         string `json:"md5"`
+	Type        string `json:"type"`
+	Version     string `json:"version"`
+}
+
+func (c *DevUpgrade) Marshal() ([]byte, error) {
+	return gjson.New(c).ToJson()
+}
+
+func (c *DevUpgrade) UnMarshal(bytes []byte) error {
+	j, err := gjson.DecodeToJson(bytes)
+	if err != nil {
+		return err
+	}
+	return j.Struct(c)
+}

pkg/mqtt/connection.go

@@ -225,17 +225,12 @@ func (c *Connection) RcvMsgFromClient() {
 				c.Close()
 				return
 			}
-
-			server.Log.Infof("device %d, connected to server now, host: %s", c.DeviceID, host)
-
 		case *Publish:
-			server.Log.Infof("%s, publish topic: %s， msgId: %v", host, msg.TopicName, msg.MessageID)
 			err = c.Mgr.PublishMessage2Server(c.DeviceCode, c.VendorId, msg)
 			if err != nil {
 				server.Log.Errorf("PublishMessage2Server error:%s", err.Error())
 			}
 			if msg.QosLevel.IsAtLeastOnce() {
-				server.Log.Infof("publish ack send now")
 				publishack := &PubAck{MessageID: msg.MessageID}
 				c.Submit(publishack)
 			} else if msg.QosLevel.IsExactlyOnce() {
@@ -252,7 +247,6 @@ func (c *Connection) RcvMsgFromClient() {
 			}
 
 		case *PubAck:
-			server.Log.Infof("%s, comes publish ack", host)
 			c.confirmPublish(msg.MessageID)
 			err := c.Mgr.Provider.OnDeviceHeartBeat(c.DeviceCode)
 			if err != nil {
@@ -282,7 +276,6 @@ func (c *Connection) RcvMsgFromClient() {
 			}
 
 		case *PingReq:
-			server.Log.Infof("%s, ping req comes", host)
 			pingrsp := &PingResp{}
 			err := c.Mgr.Provider.OnDeviceHeartBeat(c.DeviceCode)
 			if err != nil {

pkg/rpcs/access.go

@@ -27,3 +27,11 @@ type ArgsSendCommand struct {
 	Params    map[string]interface{}
 }
 type ReplySendCommand ReplyEmptyResult
+
+type ArgsDeviceUpgrade struct {
+	DeviceId    string
+	SudDeviceId string
+	Url         string
+	Md5         string
+	Version     string
+}

pkg/rpcs/device_net_config.go

@@ -14,3 +14,8 @@ type SubDeviceArgs struct {
 	SubDeviceId string `json:"sub_device_id"`
 	Status      int    `json:"status"`
 }
+
+type ArgsUpdateDeviceVersion struct {
+	DeviceId string
+	Version  string
+}

services/apiprovider/actions.go

@@ -93,11 +93,8 @@ func GetDeviceInfoByIdentifier(urlparams martini.Params, r render.Render) {
 	identifier := urlparams["identifier"]
 	server.Log.Printf("ACTION GetDeviceInfoByIdentifier, identifier:: %v", identifier)
 	device := &models.Device{}
-	span, ctx := opentracing.StartSpanFromContext(context.Background(), "GetDeviceInfoByIdentifier")
-	defer span.Finish()
-	ext.SpanKindRPCClient.Set(span)
-	span.SetTag("identifier", identifier)
-	err := server.RPCCallByName(ctx, rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", identifier, device)
+
+	err := server.RPCCallByName(context.Background(), rpcs.RegistryServerName, "Registry.FindDeviceByIdentifier", identifier, device)
 	if err != nil {
 		r.JSON(http.StatusOK, renderError(ErrDeviceNotFound, err))
 		return
@@ -149,6 +146,32 @@ func GetDeviceLatestStatus() {
 
 }
 
+// DeviceUpgrade 设备OTA升级
+func DeviceUpgrade(device *models.Device, urlparams martini.Params, req *http.Request, r render.Render) {
+	var param DeviceUpgradeReq
+	decoder := json.NewDecoder(req.Body)
+	err := decoder.Decode(&param)
+	if err != nil {
+		r.JSON(http.StatusOK, renderError(ErrWrongRequestFormat, err))
+		return
+	}
+	var args rpcs.ArgsDeviceUpgrade
+	args.DeviceId = param.DeviceId
+	args.SudDeviceId = param.SubDeviceId
+	args.Url = param.Url
+	args.Md5 = param.MD5
+	args.Version = param.Version
+	var reply rpcs.ReplyEmptyResult
+	err = server.RPCCallByName(context.Background(), rpcs.MQTTAccessName, "Access.SetStatus", args, &reply)
+	if err != nil {
+		server.Log.Errorf("设备OTA升级失败：", err)
+		r.JSON(http.StatusOK, renderError(ErrSystemFault, err))
+		return
+	}
+	r.JSON(http.StatusOK, Common{})
+	return
+}
+
 // SetDeviceStatus set device status
 func SetDeviceStatus(device *models.Device, config *productconfig.ProductConfig,
 	urlparams martini.Params, req *http.Request, r render.Render) {

services/apiprovider/request.go

@@ -6,3 +6,11 @@ type CreateRuleRequest struct {
 	Target  string `json:"target"`
 	Action  string `json:"action"`
 }
+
+type DeviceUpgradeReq struct {
+	DeviceId    string `json:"device_id"`
+	SubDeviceId string `json:"sub_device_id"`
+	Url         string `json:"url"`
+	MD5         string `json:"md5"`
+	Version     string `json:"version"`
+}

services/apiprovider/router.go

@@ -52,6 +52,8 @@ func route(m *martini.ClassicMartini) {
 			ApplicationAuthOnDeviceIdentifer, CheckDeviceOnline, CheckProductConfig,
 			SetDeviceStatus)
 
+		r.Post("/devices/:identifier/ota", ApplicationAuthOnDeviceIdentifer, CheckDeviceOnline, DeviceUpgrade)
+
 		// send a command to device
 		r.Post("/devices/:identifier/commands",
 			ApplicationAuthOnDeviceIdentifer, CheckDeviceOnline, CheckProductConfig,

services/mqttaccess/access.go

@@ -75,3 +75,23 @@ func (a *Access) SendCommand(args rpcs.ArgsSendCommand, reply *rpcs.ReplySendCom
 	}
 	return a.MqttBroker.SendMessageToDevice(args.DeviceId, "c", msg, time.Duration(args.WaitTime)*time.Second)
 }
+
+// Upgrade 设备OTA升级指令
+func (a *Access) Upgrade(args rpcs.ArgsDeviceUpgrade, reply *rpcs.ReplyEmptyResult) error {
+	server.Log.Infof("设备OTA升级：%s, %s", args.DeviceId, args.Version)
+	cmd := &klink.DevUpgrade{
+		Action:      "devUpgrade",
+		MsgId:       0,
+		DeviceCode:  args.DeviceId,
+		SubDeviceId: args.SudDeviceId,
+		Timestamp:   time.Now().Unix(),
+		Version:     args.Version,
+		Url:         args.Url,
+		Md5:         args.Md5,
+	}
+	msg, err := cmd.Marshal()
+	if err != nil {
+		return err
+	}
+	return a.MqttBroker.SendMessageToDevice(args.DeviceId, "c", msg, 5*time.Second)
+}

services/mqttaccess/mqtt_provider.go

@@ -107,6 +107,8 @@ func (mp *MQTTProvider) OnDeviceMessage(deviceid, vendorId string, msgtype strin
 				_ = processDevLogout(deviceid, message.GetString("subDeviceId"))
 			case klink.DevNetConfigAction:
 				_ = processDevNetConfig(deviceid, message.GetString("md5"))
+			case klink.ReportFirmwareAction:
+				_ = processDeviceReportUpgrade(deviceid, message.GetString("version"))
 			}
 		}
 	case "e":
@@ -185,3 +187,17 @@ func processDevNetConfig(deviceCode, md5 string) error {
 	}
 	return nil
 }
+
+// 设备上报固件信息处理
+func processDeviceReportUpgrade(deviceId, version string) error {
+	args := &rpcs.ArgsUpdateDeviceVersion{
+		DeviceId: deviceId,
+		Version:  version,
+	}
+	var reply rpcs.ReplyEmptyResult
+	err := server.RPCCallByName(nil, rpcs.RegistryServerName, "Registry.UpdateDeviceVersion", args, &reply)
+	if err != nil {
+		server.Log.Errorf("更新设备版本号失败：%v", args, err)
+	}
+	return nil
+}

services/registry/registry.go

@@ -571,3 +571,22 @@ func (r *Registry) UpdateSubDevice(args *rpcs.SubDeviceArgs, reply *models.SubDe
 	}
 	return nil
 }
+
+// UpdateDeviceVersion 更新设备版本号
+func (r *Registry) UpdateDeviceVersion(args *rpcs.ArgsUpdateDeviceVersion, reply *rpcs.ReplyEmptyResult) error {
+	db, err := getDB()
+	if err != nil {
+		return err
+	}
+	fReply := &models.Device{}
+	err = r.FindDeviceByIdentifier(args.DeviceId, fReply)
+	if err != nil {
+		return err
+	}
+	fReply.DeviceVersion = args.Version
+	err = db.Save(fReply).Error
+	if err != nil {
+		return err
+	}
+	return nil
+}