123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219 |
- package stcknsqmsg
- import (
- "encoding/json"
- "fmt"
- "net"
- "os"
- "stck/stck-nsq-msg/msg"
- "github.com/denisbrodbeck/machineid"
- "github.com/go-viper/mapstructure/v2"
- )
- type StckNsqMsg struct {
- Type string `mapstructure:"type" json:"type"`
- Data StckNsqMsgVariant `mapstructure:"data" json:"data"`
- }
- type StckNsqMsgRaw struct {
- Type string `mapstructure:"type" json:"type"`
- Data map[string]interface{} `mapstructure:"data" json:"data"`
- }
- type StckNsqMsgVariant interface {
- MsgType() string
- }
- func ToStckNsqMsg(msg StckNsqMsgVariant) StckNsqMsg {
- return StckNsqMsg{
- Type: msg.MsgType(),
- Data: msg,
- }
- }
- func ToStckNsqMsgString(msg StckNsqMsgVariant) (string, error) {
- m := ToStckNsqMsg(msg)
- return m.ToString()
- }
- func (m *StckNsqMsg) ToString() (string, error) {
- var rm StckNsqMsgRaw
- rm.Type = m.Type
- // Cast to generic map
- err := mapstructure.Decode(m.Data, &rm.Data)
- if err != nil {
- return "", err
- }
- b, err := json.Marshal(rm)
- if err != nil {
- return "", err
- }
- return string(b), nil
- }
- func FromString(s string) (StckNsqMsg, error) {
- var rm StckNsqMsgRaw
- var m StckNsqMsg
- err := json.Unmarshal([]byte(s), &rm)
- if err != nil {
- return m, err
- }
- // Cast to strongly typed struct
- m.Type = rm.Type
- switch rm.Type {
- case msg.DeviceHeartbeatMsgType:
- var d msg.DeviceHeartbeatMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.DeviceActionDoneMsgType:
- var d msg.DeviceActionDoneMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.DevicePongMsgType:
- var d msg.DevicePongMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.PartUploadStatusChangedMsgType:
- var d msg.PartUploadStatusChangedMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.PartInsertToStckStatusChangedMsgType:
- var d msg.PartInsertToStckStatusChangedMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.StreamUploadStatusChangedMsgType:
- var d msg.StreamUploadStatusChangedMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.StreamInsertToStckStatusChangedMsgType:
- var d msg.StreamInsertToStckStatusChangedMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.ForceDeviceRebootMsgType:
- var d msg.ForceDeviceRebootMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.ForceDeviceSendHeartbeatMsgType:
- var d msg.ForceDeviceSendHeartbeatMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.RequestDeviceUpdateMsgType:
- var d msg.RequestDeviceUpdateMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.RequestDeviceUploadLogsToMinioMsgType:
- var d msg.RequestDeviceUploadLogsToMinioMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.DevicePingMsgType:
- var d msg.DevicePingMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- case msg.RequestDeviceExecuteShellScriptMsgType:
- var d msg.RequestDeviceExecuteShellScriptMsg
- err = mapstructure.Decode(rm.Data, &d)
- if err != nil {
- return m, err
- }
- m.Data = &d
- default:
- return m, fmt.Errorf("unknown message type: %s", rm.Type)
- }
- return m, nil
- }
- func DeviceIdMatches(id1, id2 string) bool {
- return id1 == id2 || id1 == "*" || id2 == "*"
- }
- func GetLocalIP() (net.IP, error) {
- ifaces, err := net.Interfaces()
- if err != nil {
- return nil, err
- }
- for _, iface := range ifaces {
- addrs, err := iface.Addrs()
- if err != nil {
- return nil, err
- }
- for _, addr := range addrs {
- var ip net.IP
- switch v := addr.(type) {
- case *net.IPNet:
- ip = v.IP
- case *net.IPAddr:
- ip = v.IP
- }
- if ip == nil || ip.IsLoopback() {
- continue
- }
- ip = ip.To4()
- if ip == nil {
- continue // not an ipv4 address
- }
- return ip, nil
- }
- }
- return nil, fmt.Errorf("no non-loopback IP address found")
- }
- // Generate a unique machine identifier based on the local IP and OS machine ID
- func MakeUniqueMachineID() string {
- ip, err := GetLocalIP()
- if err != nil {
- fmt.Fprintf(os.Stderr, "get local IP failed: %v\n", err)
- }
- machineid, err := machineid.ProtectedID("unique")
- if err != nil {
- fmt.Fprintf(os.Stderr, "get machine ID failed: %v\n", err)
- }
- if len(machineid) > 48 {
- machineid = machineid[:48] // truncate to 48 characters
- }
- return fmt.Sprintf("%s-%s", ip.String(), machineid)
- }
|