// Device is the Schema for the devices API
  // +k8s:openapi-gen=true
  type Device struct {
  	metav1.TypeMeta   `json:",inline"`
  	metav1.ObjectMeta `json:"metadata,omitempty"`
  
  	Spec   DeviceSpec   `json:"spec,omitempty"`
  	Status DeviceStatus `json:"status,omitempty"`
  }
  // DeviceSpec represents a single device instance. It is an instantation of a device model.
  type DeviceSpec struct {
  	// Required: DeviceModelRef is reference to the device model used as a template
  	// to create the device instance.
  	DeviceModelRef *v1.LocalObjectReference `json:"deviceModelRef,omitempty"`
  	// Required: The protocol configuration used to connect to the device.
  	Protocol ProtocolConfig `json:"protocol,omitempty"`
  	// List of property visitors which describe how to access the device properties.
  	// PropertyVisitors must unique by propertyVisitor.propertyName.
  	// +optional
  	PropertyVisitors []DevicePropertyVisitor `json:"propertyVisitors,omitempty"`
  	// Data section describe a list of time-series properties which should be processed
  	// on edge node.
  	// +optional
  	Data DeviceData `json:"data,omitempty"`
  	// NodeSelector indicates the binding preferences between devices and nodes.
  	// Refer to k8s.io/kubernetes/pkg/apis/core NodeSelector for more details
  	// +optional
  	NodeSelector *v1.NodeSelector `json:"nodeSelector,omitempty"`
  }
  // DeviceStatus reports the device state and the desired/reported values of twin attributes.
  type DeviceStatus struct {
  	// A list of device twins containing desired/reported desired/reported values of twin properties..
  	// Optional: A passive device won't have twin properties and this list could be empty.
  	// +optional
  	Twins []Twin `json:"twins,omitempty"`
  }

  //Device the struct of device
  type Device struct {
  	ID          string              `json:"id,omitempty"`
  	Name        string              `json:"name,omitempty"`
  	Description string              `json:"description,omitempty"`
  	State       string              `json:"state,omitempty"`
  	LastOnline  string              `json:"last_online,omitempty"`
  	Attributes  map[string]*MsgAttr `json:"attributes,omitempty"`
  	Twin        map[string]*MsgTwin `json:"twin,omitempty"`
  }
  //MsgAttr the struct of device attr
  type MsgAttr struct {
  	Value    string        `json:"value"`
  	Optional *bool         `json:"optional,omitempty"`
  	Metadata *TypeMetadata `json:"metadata,omitempty"`
  }
  //MsgTwin the struct of device twin
  type MsgTwin struct {
  	Expected        *TwinValue    `json:"expected,omitempty"`
  	Actual          *TwinValue    `json:"actual,omitempty"`
  	Optional        *bool         `json:"optional,omitempty"`
  	Metadata        *TypeMetadata `json:"metadata,omitempty"`
  	ExpectedVersion *TwinVersion  `json:"expected_version,omitempty"`
  	ActualVersion   *TwinVersion  `json:"actual_version,omitempty"`
  }

  //DTMessage the struct of message for communicating between cloud and edge
  type DTMessage struct {
  	Msg      *model.Message
  	Identity string //表示节点ID
  	Action   string //代表该条消息执行的动作，根据不同的Action将消息转发至对应的DeviceTwin子模块
  	Type     string //这个貌似也没啥用
  }

// Start run the module
func (dt *DeviceTwin) Start() {
	dtContexts, _ := dtcontext.InitDTContext()
	dt.DTContexts = dtContexts
  err := SyncSqlite(dt.DTContexts) //将存储在本地数据库中的device信息载入DTContexts
	...
	dt.runDeviceTwin() //运行各个子模块
}

// SyncSqlite sync sqlite
func SyncSqlite(context *dtcontext.DTContext) error {
	...
	rows, queryErr := dtclient.QueryDeviceAll() //从本地数据库中查询所有设备
	...
	for _, device := range *rows {
		err := SyncDeviceFromSqlite(context, device.ID) //将设备保存至DTContext中
		...
	}
	return nil
}

func (dt *DeviceTwin) runDeviceTwin() {
	moduleNames := []string{dtcommon.MemModule, dtcommon.TwinModule, dtcommon.DeviceModule, dtcommon.CommModule}
  //1.注册并启动子模块
	for _, v := range moduleNames {
		dt.RegisterDTModule(v)
		go dt.DTModules[v].Start()
	}
	go func() {
		for {
			...
			if msg, ok := beehiveContext.Receive("twin"); ok == nil {
				klog.Info("DeviceTwin receive msg")
        //2.消息转发
				err := dt.distributeMsg(msg)
				if err != nil {
					klog.Warningf("distributeMsg failed: %v", err)
				}
			}
		}
	}()

  //3.健康检查，如有必要则会重启子模块
	for {
		select {
		case <-time.After((time.Duration)(60) * time.Second):
			//range to check whether has bug
			for dtmName := range dt.DTModules {
				health, ok := dt.DTContexts.ModulesHealth.Load(dtmName)
				if ok {
					now := time.Now().Unix()
					if now-health.(int64) > 60*2 {
						klog.Infof("%s health %v is old, and begin restart", dtmName, health)
						go dt.DTModules[dtmName].Start()
					}
				}
			}
			for _, v := range dt.HeartBeatToModule {
				v <- "ping"
			}
		case <-beehiveContext.Done():
			for _, v := range dt.HeartBeatToModule {
				v <- "stop"
			}
			klog.Warning("Stop DeviceTwin ModulesHealth load loop")
			return
		}
	}
}

//distributeMsg distribute message to diff module
func (dt *DeviceTwin) distributeMsg(m interface{}) error {
   msg, ok := m.(model.Message)
   ...
   message := dttype.DTMessage{Msg: &msg}
   //如果msg的parentID不为空，说明该消息是一个确认消息(response message)
   //如果是云端下发的消息，parentID均为空
   if message.Msg.GetParentID() != "" {
      klog.Infof("Send msg to the %s module in twin", dtcommon.CommModule)
      confirmMsg := dttype.DTMessage{Msg: model.NewMessage(message.Msg.GetParentID()), 
                                     Action: dtcommon.Confirm}
      if err := dt.DTContexts.CommTo(dtcommon.CommModule, &confirmMsg); err != nil {
         return err
      }
   }
   //根据msg的source和resource type进行分类
   //并填充message
   if !classifyMsg(&message) { //关键~
      return errors.New("Not found action")
   }
   if ActionModuleMap == nil {
      initActionModuleMap()
   }
	
   //根据消息的Action，将消息转发至对应的子模块中
   if moduleName, exist := ActionModuleMap[message.Action]; exist {
   		...	
      if err := dt.DTContexts.CommTo(moduleName, &message); err != nil {
         return err
      }
   } 
   ...
   return nil
}

func classifyMsg(message *dttype.DTMessage) bool {
	if EventActionMap == nil {
		initEventActionMap()
	}
	var identity string
	var action string
	msgSource := message.Msg.GetSource()
	if strings.Compare(msgSource, "bus") == 0 { //如果消息源来自”eventbus“
		...
	}
  //如果消息源来自"edgemgr"或"devicecontroller"
  else if (strings.Compare(msgSource, "edgemgr") == 0) || 
            (strings.Compare(msgSource, "devicecontroller") == 0) {
		...
    //根据message.Msg.Router.Resource进行分类
		if strings.Contains(message.Msg.Router.Resource, "membership/detail") {
			message.Action = dtcommon.MemDetailResult
			return true
		} else if strings.Contains(message.Msg.Router.Resource, "membership") { // 标记1
			message.Action = dtcommon.MemUpdated
			return true
		} else if strings.Contains(message.Msg.Router.Resource, "twin/cloud_updated") {
			message.Action = dtcommon.TwinCloudSync
			resources := strings.Split(message.Msg.Router.Resource, "/")
			message.Identity = resources[1] //nodeID
			return true
		} else if strings.Contains(message.Msg.Router.Operation, "updated") {
			resources := strings.Split(message.Msg.Router.Resource, "/")
			if len(resources) == 2 && strings.Compare(resources[0], "device") == 0 {
				message.Action = dtcommon.DeviceUpdated
				message.Identity = resources[1]
			}
			return true
		}
		return false
	} else if strings.Compare(msgSource, "edgehub") == 0 {
		...
}

//cloud/pkg/devicecontroller/controller/downstream.go -> deviceAdded()
msg := model.NewMessage("")
resource, err := messagelayer.BuildResource(device.Spec.NodeSelector.NodeSelectorTerms[0].MatchExpressions[0].Values[0], "membership", "")
msg.BuildRouter(modules.DeviceControllerModuleName, constants.GroupTwin, resource, model.UpdateOperation)
msg.Content = xxx //这里存放消息传递的有效数据

msg.Router.Source = "devicecontroller"  //表示消息源
msg.Router.Group = "twin"               //表示消息发送的组别
msg.Router.Resource = "node/edgenode_1/membership" //表示消息的资源类型，也表示topic类型
msg.Router.Operation = "update"

//Start worker
func (mw MemWorker) Start() {
	initMemActionCallBack() //初始化DTMessage.Action与回调函数的映射关系
	for {
		select {
		case msg, ok := <-mw.ReceiverChan:
			if !ok {
				return
			}
			if dtMsg, isDTMessage := msg.(*dttype.DTMessage); isDTMessage {
        //根据Action，调用对应的callback函数
				if fn, exist := memActionCallBack[dtMsg.Action]; exist { 
					_, err := fn(mw.DTContexts, dtMsg.Identity, dtMsg.Msg)
					...
				} 
			}

		case v, ok := <-mw.HeartBeatChan:
			...
			if err := mw.DTContexts.HeartBeat(mw.Group, v); err != nil {
				return
			}
		}
	}
}