Skip to content

Commit b956d61

Browse files
author
xiejingru
authored
bugfix of device management (#247)
1 parent 6784c3b commit b956d61

File tree

3 files changed

+145
-98
lines changed

3 files changed

+145
-98
lines changed

‎dmcontext/config.go

Lines changed: 47 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,10 @@
11
package dmcontext
22

33
import (
4-
"github.com/baetyl/baetyl-go/v2/context"
54
mqtt2 "github.com/baetyl/baetyl-go/v2/mqtt"
5+
v1 "github.com/baetyl/baetyl-go/v2/spec/v1"
66
)
77

8-
type SystemConfig struct {
9-
context.SystemConfig `yaml:",inline" json:",inline"`
10-
Devices []DeviceInfo `yaml:"devices,omitempty" json:"devices,omitempty"`
11-
}
12-
138
type DeviceInfo struct {
149
Name string `yaml:"name,omitempty" json:"name,omitempty"`
1510
Version string `yaml:"version,omitempty" json:"version,omitempty"`
@@ -23,3 +18,49 @@ type Topic struct {
2318
Get mqtt2.QOSTopic `yaml:"get,omitempty" json:"get,omitempty"`
2419
GetResponse mqtt2.QOSTopic `yaml:"getResponse,omitempty" json:"getResponse,omitempty"`
2520
}
21+
22+
type DeviceProperty struct {
23+
Name string `yaml:"name,omitempty" json:"name,omitempty"`
24+
Type string `yaml:"type,omitempty" json:"type,omitempty" validate:"regexp=^(int16|int32|int64|float32|float64|string|bool)?$"`
25+
Mode string `yaml:"mode,omitempty" json:"mode,omitempty" validate:"regexp=^(ro|rw)?$"`
26+
Visitor PropertyVisitor `yaml:"visitor,omitempty" json:"visitor,omitempty"`
27+
}
28+
29+
type PropertyVisitor struct {
30+
Modbus *ModbusVisitor `yaml:"modbus,omitempty" json:"modbus,omitempty"`
31+
Opcua *OpcuaVisitor `yaml:"opcua,omitempty" json:"opcua,omitempty"`
32+
Custom *CustomVisitor `yaml:"custom,omitempty" json:"custom,omitempty"`
33+
}
34+
35+
type ModbusVisitor struct {
36+
Function byte `yaml:"function" json:"function" validate:"min=1,max=4"`
37+
Address string `yaml:"address" json:"address"`
38+
Quantity uint16 `yaml:"quantity" json:"quantity"`
39+
Type string `yaml:"type,omitempty" json:"type,omitempty" validate:"regexp=^(int16|int32|int64|float32|float64|string|bool)?$"`
40+
Scale float64 `yaml:"scale" json:"scale"`
41+
SwapByte bool `yaml:"swapByte" json:"swapByte"`
42+
SwapRegister bool `yaml:"swapRegister" json:"swapRegister"`
43+
}
44+
45+
type OpcuaVisitor struct {
46+
NodeID string `yaml:"nodeid,omitempty" json:"nodeid,omitempty"`
47+
Type string `yaml:"type,omitempty" json:"type,omitempty" validate:"regexp=^(int16|int32|int64|float32|float64|string|bool)?$"`
48+
}
49+
50+
type CustomVisitor string
51+
52+
type Event struct {
53+
Type string `yaml:"type,omitempty" json:"type,omitempty"`
54+
Payload interface{} `yaml:"payload,omitempty" json:"payload,omitempty"`
55+
}
56+
57+
type DeviceShadow struct {
58+
Name string `yaml:"name,omitempty" json:"name,omitempty"`
59+
Report v1.Report `yaml:"report,omitempty" json:"report,omitempty"`
60+
Desire v1.Desire `yaml:"desire,omitempty" json:"desire,omitempty"`
61+
}
62+
63+
type driverConfig struct {
64+
Devices []DeviceInfo `yaml:"devices,omitempty" json:"devices,omitempty"`
65+
Driver string `yaml:"driver,omitempty" json:"driver,omitempty"`
66+
}

‎dmcontext/context.go

Lines changed: 96 additions & 90 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,8 @@ import (
77
"sync"
88
"time"
99

10+
"gopkg.in/yaml.v2"
11+
1012
"github.com/baetyl/baetyl-go/v2/context"
1113
"github.com/baetyl/baetyl-go/v2/errors"
1214
"github.com/baetyl/baetyl-go/v2/log"
@@ -16,92 +18,58 @@ import (
1618
)
1719

1820
const (
19-
DefaultAccessConf = "/etc/baetyl/access.yml"
20-
DefaultPropsConf = "/etc/baetyl/props.yml"
21+
DefaultAccessConf = "etc/baetyl/access.yml"
22+
DefaultPropsConf = "etc/baetyl/props.yml"
23+
DefaultDriverConf = "etc/baetyl/conf.yml"
2124
KeyDevice = "device"
2225
KeyStatus = "status"
2326
OnlineStatus = "online"
2427
OfflineStatus = "offline"
2528
TypeReportEvent = "report"
26-
KeySysExtConf = "BAETYL_SYSTEM_EXT_CONF"
2729
)
2830

2931
var (
3032
ErrInvalidMessage = errors.New("invalid device message")
3133
ErrInvalidChannel = errors.New("invalid channel")
3234
ErrResponseChannelNotExist = errors.New("response channel not exist")
35+
ErrAccessConfigNotExist = errors.New("access config not exist")
36+
ErrPropsConfigNotExist = errors.New("properties config not exist")
3337
)
3438

35-
type DeviceProperty struct {
36-
Name string `json:"name,omitempty"`
37-
Type string `json:"type,omitempty" validate:"regexp=^(int16|int32|int64|float32|float64|string|bool)?$"`
38-
Mode string `json:"mode,omitempty" validate:"regexp=^(ro|rw)?$"`
39-
Visitor PropertyVisitor `json:"visitor,omitempty"`
40-
}
41-
42-
type PropertyVisitor struct {
43-
Modbus *ModbusVisitor `json:"modbus,omitempty"`
44-
Opcua *OpcuaVisitor `json:"opcua,omitempty"`
45-
Custom *CustomVisitor `json:"custom,omitempty"`
46-
}
47-
48-
type ModbusVisitor struct {
49-
Function byte `json:"function" validate:"min=1,max=4"`
50-
Address string `json:"address"`
51-
Quantity uint16 `json:"quantity"`
52-
Type string `json:"type,omitempty" validate:"regexp=^(int16|int32|int64|float32|float64|string|bool)?$"`
53-
Scale float64 `json:"scale"`
54-
SwapByte bool `json:"swapByte"`
55-
SwapRegister bool `json:"swapRegister"`
56-
}
57-
58-
type OpcuaVisitor struct {
59-
NodeID string `json:"nodeid,omitempty"`
60-
Type string `json:"type,omitempty" validate:"regexp=^(int16|int32|int64|float32|float64|string|bool)?$"`
61-
}
62-
63-
type CustomVisitor string
64-
65-
type Event struct {
66-
Type string `yaml:"type,omitempty" json:"type,omitempty"`
67-
Payload interface{} `yaml:"payload,omitempty" json:"payload,omitempty"`
68-
}
69-
70-
type DeviceShadow struct {
71-
Name string `json:"name,omitempty"`
72-
Report v1.Report `json:"report,omitempty"`
73-
Desire v1.Desire `json:"desire,omitempty"`
74-
}
75-
7639
type DeltaCallback func(*DeviceInfo, v1.Delta) error
77-
7840
type EventCallback func(*DeviceInfo, *Event) error
7941

8042
type Context interface {
8143
context.Context
82-
SystemConfigExt() *SystemConfig
8344
GetAllDevices() []DeviceInfo
8445
ReportDeviceProperties(*DeviceInfo, v1.Report) error
85-
GetDeviceProperties(info *DeviceInfo) (*DeviceShadow, error)
46+
GetDeviceProperties(device *DeviceInfo) (*DeviceShadow, error)
8647
RegisterDeltaCallback(cb DeltaCallback) error
8748
RegisterEventCallback(cb EventCallback) error
88-
Online(info *DeviceInfo) error
89-
Offline(info *DeviceInfo) error
90-
GetDeviceAccessConfig() (string, error)
91-
GetDevicePropConfigs() (map[string][]DeviceProperty, error)
49+
Online(device *DeviceInfo) error
50+
Offline(device *DeviceInfo) error
51+
GetDriverConfig() string
52+
GetAccessConfig() map[string]string
53+
GetDeviceAccessConfig(device *DeviceInfo) (string, error)
54+
GetPropertiesConfig() map[string][]DeviceProperty
55+
GetDevicePropertiesConfig(device *DeviceInfo) ([]DeviceProperty, error)
9256
Start()
9357
io.Closer
9458
}
9559

9660
type DmCtx struct {
9761
context.Context
98-
log *log.Logger
99-
mqtt *mqtt2.Client
100-
tomb utils.Tomb
101-
eventCb EventCallback
102-
deltaCb DeltaCallback
103-
response sync.Map
104-
msgChs map[string]chan *v1.Message
62+
log *log.Logger
63+
mqtt *mqtt2.Client
64+
tomb utils.Tomb
65+
eventCb EventCallback
66+
deltaCb DeltaCallback
67+
response sync.Map
68+
devices map[string]DeviceInfo
69+
msgChs map[string]chan *v1.Message
70+
driverConfig string
71+
propsConfig map[string][]DeviceProperty
72+
accessConfig map[string]string
10573
}
10674

10775
func NewContext(confFile string) Context {
@@ -119,17 +87,31 @@ func NewContext(confFile string) Context {
11987
lfs = append(lfs, log.Any("service", c.ServiceName()))
12088
}
12189
c.log = log.With(lfs...)
122-
sc := new(SystemConfig)
123-
if err := c.LoadCustomConfig(sc); err != nil {
124-
c.log.Error("failed to load system config, to use default config", log.Error(err))
125-
utils.UnmarshalYAML(nil, sc)
90+
91+
if err := unmarshalYAML(DefaultAccessConf, &c.accessConfig); err != nil {
92+
c.log.Error("failed to load access config, to use default config", log.Error(err))
93+
utils.UnmarshalYAML(nil, &c.accessConfig)
94+
}
95+
96+
if err := unmarshalYAML(DefaultPropsConf, &c.propsConfig); err != nil {
97+
c.log.Error("failed to load props config, to use default config", log.Error(err))
98+
utils.UnmarshalYAML(nil, &c.propsConfig)
12699
}
127-
c.Store(KeySysExtConf, sc)
128100

101+
var dCfg driverConfig
102+
if err := unmarshalYAML(DefaultDriverConf, &dCfg); err != nil {
103+
c.log.Error("failed to load driver config, to use default config", log.Error(err))
104+
utils.UnmarshalYAML(nil, &dCfg)
105+
}
106+
c.driverConfig = dCfg.Driver
107+
108+
devices := make(map[string]DeviceInfo)
129109
var subs []mqtt2.QOSTopic
130-
for _, dev := range sc.Devices {
110+
for _, dev := range dCfg.Devices {
131111
subs = append(subs, dev.Delta, dev.Event, dev.GetResponse)
112+
devices[dev.Name] = dev
132113
}
114+
c.devices = devices
133115
mqtt, err := c.Context.NewSystemBrokerClient(subs)
134116
if err != nil {
135117
c.log.Warn("fail to create system broker client", log.Any("error", err))
@@ -143,9 +125,8 @@ func NewContext(confFile string) Context {
143125
}
144126

145127
func (c *DmCtx) Start() {
146-
devices := c.SystemConfigExt().Devices
147-
for _, dev := range devices {
148-
c.msgChs[dev.Name] = make(chan *v1.Message, 1024)
128+
for name, dev := range c.devices {
129+
c.msgChs[name] = make(chan *v1.Message, 1024)
149130
go c.processing(c.msgChs[dev.Name])
150131
}
151132
}
@@ -159,7 +140,7 @@ func (c *DmCtx) Close() error {
159140
}
160141

161142
func (c *DmCtx) processDelta(msg *v1.Message) error {
162-
device := msg.Metadata[KeyDevice]
143+
deviceName := msg.Metadata[KeyDevice]
163144
if c.deltaCb == nil {
164145
c.log.Debug("delta callback not set and message will not be process")
165146
return nil
@@ -168,14 +149,19 @@ func (c *DmCtx) processDelta(msg *v1.Message) error {
168149
if err := msg.Content.Unmarshal(&delta); err != nil {
169150
return errors.Trace(err)
170151
}
171-
if err := c.deltaCb(&DeviceInfo{Name: device}, delta); err != nil {
152+
dev, ok := c.devices[deviceName]
153+
if !ok {
154+
c.log.Warn("delta callback can not find device", log.Any("device", deviceName))
155+
return nil
156+
}
157+
if err := c.deltaCb(&dev, delta); err != nil {
172158
return errors.Trace(err)
173159
}
174160
return nil
175161
}
176162

177163
func (c *DmCtx) processEvent(msg *v1.Message) error {
178-
device := msg.Metadata[KeyDevice]
164+
deviceName := msg.Metadata[KeyDevice]
179165
if c.eventCb == nil {
180166
c.log.Debug("event callback not set and message will not be process")
181167
return nil
@@ -184,7 +170,12 @@ func (c *DmCtx) processEvent(msg *v1.Message) error {
184170
if err := msg.Content.Unmarshal(&event); err != nil {
185171
return errors.Trace(err)
186172
}
187-
if err := c.eventCb(&DeviceInfo{Name: device}, &event); err != nil {
173+
dev, ok := c.devices[deviceName]
174+
if !ok {
175+
c.log.Warn("event callback can not find device", log.Any("device", deviceName))
176+
return nil
177+
}
178+
if err := c.eventCb(&dev, &event); err != nil {
188179
return errors.Trace(err)
189180
}
190181
return nil
@@ -247,16 +238,12 @@ func (c *DmCtx) processing(ch chan *v1.Message) {
247238
}
248239
}
249240

250-
func (c *DmCtx) SystemConfigExt() *SystemConfig {
251-
v, ok := c.Load(KeySysExtConf)
252-
if !ok {
253-
return nil
254-
}
255-
return v.(*SystemConfig)
256-
}
257-
258241
func (c *DmCtx) GetAllDevices() []DeviceInfo {
259-
return c.SystemConfigExt().Devices
242+
var deviceList []DeviceInfo
243+
for _, dev := range c.devices {
244+
deviceList = append(deviceList, dev)
245+
}
246+
return deviceList
260247
}
261248

262249
func (c *DmCtx) ReportDeviceProperties(info *DeviceInfo, report v1.Report) error {
@@ -361,18 +348,37 @@ func (c *DmCtx) Offline(info *DeviceInfo) error {
361348
return nil
362349
}
363350

364-
func (c *DmCtx) GetDeviceAccessConfig() (string, error) {
365-
res, err := ioutil.ReadFile(DefaultAccessConf)
366-
if err != nil {
367-
return "", errors.Trace(err)
351+
func (c *DmCtx) GetDriverConfig() string {
352+
return c.driverConfig
353+
}
354+
func (c *DmCtx) GetAccessConfig() map[string]string {
355+
return c.accessConfig
356+
}
357+
358+
func (c *DmCtx) GetDeviceAccessConfig(device *DeviceInfo) (string, error) {
359+
if cfg, ok := c.accessConfig[device.Name]; ok {
360+
return cfg, nil
361+
} else {
362+
return "", ErrAccessConfigNotExist
368363
}
369-
return string(res), nil
370364
}
371365

372-
func (c *DmCtx) GetDevicePropConfigs() (map[string][]DeviceProperty, error) {
373-
var res map[string][]DeviceProperty
374-
if err := c.LoadCustomConfig(&res, DefaultPropsConf); err != nil {
375-
return nil, errors.Trace(err)
366+
func (c *DmCtx) GetPropertiesConfig() map[string][]DeviceProperty {
367+
return c.propsConfig
368+
}
369+
370+
func (c *DmCtx) GetDevicePropertiesConfig(device *DeviceInfo) ([]DeviceProperty, error) {
371+
if cfg, ok := c.propsConfig[device.Name]; ok {
372+
return cfg, nil
373+
} else {
374+
return nil, ErrPropsConfigNotExist
375+
}
376+
}
377+
378+
func unmarshalYAML(file string, out interface{}) error {
379+
bs, err := ioutil.ReadFile(file)
380+
if err != nil {
381+
return err
376382
}
377-
return res, nil
383+
return yaml.Unmarshal(bs, out)
378384
}

‎spec/v1/lazy_value_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -82,8 +82,8 @@ func TestSpecV1_LazyValue(t *testing.T) {
8282
s := "test"
8383
b := true
8484
msg4 := &Message{
85-
Kind: MessageReport,
86-
Content: LazyValue{Value: map[string]interface{}{"int64": i64, "string": s, "bool": b}},
85+
Kind: MessageReport,
86+
Content: LazyValue{Value: map[string]interface{}{"int64": i64, "string": s, "bool": b}},
8787
}
8888
data5, err := json.Marshal(msg4)
8989
assert.NoError(t, err)

0 commit comments

Comments
 (0)