feat: batch device property reads
This commit is contained in:
@@ -22,6 +22,7 @@ const (
|
|||||||
deviceSpecUA = "mijiaAPI/4.1.2"
|
deviceSpecUA = "mijiaAPI/4.1.2"
|
||||||
deviceSpecMaxSize = 8 << 20
|
deviceSpecMaxSize = 8 << 20
|
||||||
deviceCacheVersion = 2
|
deviceCacheVersion = 2
|
||||||
|
deviceGetBatchSize = 20
|
||||||
)
|
)
|
||||||
|
|
||||||
var deviceSpecURL = "https://home.miot-spec.com/spec/"
|
var deviceSpecURL = "https://home.miot-spec.com/spec/"
|
||||||
@@ -705,6 +706,76 @@ func (device *Device) Get(ctx context.Context, name string) (any, error) {
|
|||||||
return results[0].Value, nil
|
return results[0].Value, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (device *Device) GetMany(ctx context.Context, names []string) ([]DevicePropertyResult, error) {
|
||||||
|
if len(names) == 0 {
|
||||||
|
return []DevicePropertyResult{}, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
type propertyIdentity struct {
|
||||||
|
did string
|
||||||
|
siid int
|
||||||
|
piid int
|
||||||
|
}
|
||||||
|
requests := make([]PropertyRequest, len(names))
|
||||||
|
seenNames := make(map[string]struct{}, len(names))
|
||||||
|
seenIdentities := make(map[propertyIdentity]struct{}, len(names))
|
||||||
|
for index, name := range names {
|
||||||
|
if _, duplicate := seenNames[name]; duplicate {
|
||||||
|
return nil, fmt.Errorf("重复的属性: %s", name)
|
||||||
|
}
|
||||||
|
seenNames[name] = struct{}{}
|
||||||
|
property, ok := device.properties[name]
|
||||||
|
if !ok {
|
||||||
|
return nil, fmt.Errorf("不支持的属性: %s", name)
|
||||||
|
}
|
||||||
|
if !strings.Contains(property.RW, "r") {
|
||||||
|
return nil, fmt.Errorf("属性 %s 不可读取", name)
|
||||||
|
}
|
||||||
|
identity := propertyIdentity{did: device.DID, siid: property.SIID, piid: property.PIID}
|
||||||
|
if _, duplicate := seenIdentities[identity]; duplicate {
|
||||||
|
return nil, fmt.Errorf("属性 %s 与其他请求使用重复的设备属性 identity", name)
|
||||||
|
}
|
||||||
|
seenIdentities[identity] = struct{}{}
|
||||||
|
requests[index] = PropertyRequest{DID: identity.did, SIID: identity.siid, PIID: identity.piid}
|
||||||
|
}
|
||||||
|
|
||||||
|
results := make([]DevicePropertyResult, len(names))
|
||||||
|
for start := 0; start < len(requests); start += deviceGetBatchSize {
|
||||||
|
end := min(start+deviceGetBatchSize, len(requests))
|
||||||
|
chunkResults, err := device.client.GetProperties(ctx, requests[start:end])
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
byIdentity := make(map[propertyIdentity]PropertyResult, len(chunkResults))
|
||||||
|
requested := make(map[propertyIdentity]struct{}, end-start)
|
||||||
|
for _, request := range requests[start:end] {
|
||||||
|
requested[propertyIdentity{did: request.DID, siid: request.SIID, piid: request.PIID}] = struct{}{}
|
||||||
|
}
|
||||||
|
for _, result := range chunkResults {
|
||||||
|
identity := propertyIdentity{did: result.DID, siid: result.SIID, piid: result.PIID}
|
||||||
|
if _, expected := requested[identity]; !expected {
|
||||||
|
return nil, fmt.Errorf("get properties protocol error: unexpected identity (%s,%d,%d)", result.DID, result.SIID, result.PIID)
|
||||||
|
}
|
||||||
|
if _, duplicate := byIdentity[identity]; duplicate {
|
||||||
|
return nil, fmt.Errorf("get properties protocol error: duplicate identity (%s,%d,%d)", result.DID, result.SIID, result.PIID)
|
||||||
|
}
|
||||||
|
byIdentity[identity] = result
|
||||||
|
}
|
||||||
|
for index, request := range requests[start:end] {
|
||||||
|
identity := propertyIdentity{did: request.DID, siid: request.SIID, piid: request.PIID}
|
||||||
|
result, ok := byIdentity[identity]
|
||||||
|
if !ok {
|
||||||
|
return nil, fmt.Errorf("get properties protocol error: missing identity (%s,%d,%d)", request.DID, request.SIID, request.PIID)
|
||||||
|
}
|
||||||
|
results[start+index] = DevicePropertyResult{Name: names[start+index], Value: result.Value, Code: result.Code}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if err := device.wait(ctx); err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
return results, nil
|
||||||
|
}
|
||||||
|
|
||||||
func (device *Device) Set(ctx context.Context, name string, value any) error {
|
func (device *Device) Set(ctx context.Context, name string, value any) error {
|
||||||
property, ok := device.properties[name]
|
property, ok := device.properties[name]
|
||||||
if !ok {
|
if !ok {
|
||||||
|
|||||||
+134
@@ -569,6 +569,140 @@ func TestDeviceGetSetAndAction(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestDeviceGetManyChunksProperties(t *testing.T) {
|
||||||
|
for _, count := range []int{20, 21} {
|
||||||
|
t.Run(fmt.Sprint(count), func(t *testing.T) {
|
||||||
|
properties, names := batchPropertyFixture(count)
|
||||||
|
responses := make([]string, 0, (count+19)/20)
|
||||||
|
for start := 0; start < count; start += 20 {
|
||||||
|
end := min(start+20, count)
|
||||||
|
items := make([]PropertyResult, 0, end-start)
|
||||||
|
for index := start; index < end; index++ {
|
||||||
|
property := properties[names[index]]
|
||||||
|
items = append(items, PropertyResult{DID: "a", SIID: property.SIID, PIID: property.PIID, Value: index, Code: 0})
|
||||||
|
}
|
||||||
|
payload, err := json.Marshal(items)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
responses = append(responses, string(payload))
|
||||||
|
}
|
||||||
|
device, testServer := fixtureDeviceWithServer(t, responses)
|
||||||
|
device.properties = properties
|
||||||
|
|
||||||
|
results, err := device.GetMany(context.Background(), names)
|
||||||
|
if err != nil || len(results) != count {
|
||||||
|
t.Fatalf("GetMany() = %#v, %v", results, err)
|
||||||
|
}
|
||||||
|
wantCalls := (count + 19) / 20
|
||||||
|
if got := len(testServer.requests) - 2; got != wantCalls {
|
||||||
|
t.Fatalf("property calls = %d, want %d", got, wantCalls)
|
||||||
|
}
|
||||||
|
for index, request := range testServer.requests[2:] {
|
||||||
|
params := request["params"].([]any)
|
||||||
|
wantSize := min(20, count-index*20)
|
||||||
|
if len(params) != wantSize {
|
||||||
|
t.Fatalf("chunk %d size = %d, want %d", index, len(params), wantSize)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeviceGetManyMatchesIdentityAndPreservesBusinessErrors(t *testing.T) {
|
||||||
|
device := fixtureDeviceWithResults(t, []string{`[
|
||||||
|
{"did":"a","siid":2,"piid":2,"code":-704030013},
|
||||||
|
{"did":"a","siid":2,"piid":1,"value":true,"code":0}
|
||||||
|
]`}, 0)
|
||||||
|
|
||||||
|
results, err := device.GetMany(context.Background(), []string{"power", "brightness"})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
want := []DevicePropertyResult{{Name: "power", Value: true, Code: 0}, {Name: "brightness", Code: -704030013}}
|
||||||
|
if !reflect.DeepEqual(results, want) {
|
||||||
|
t.Fatalf("GetMany() = %#v, want %#v", results, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeviceGetManyRejectsInvalidResponseIdentities(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
response string
|
||||||
|
}{
|
||||||
|
{name: "missing", response: `[{"did":"a","siid":2,"piid":1,"value":true,"code":0}]`},
|
||||||
|
{name: "duplicate", response: `[{"did":"a","siid":2,"piid":1,"value":true,"code":0},{"did":"a","siid":2,"piid":1,"value":false,"code":0}]`},
|
||||||
|
{name: "extra", response: `[{"did":"a","siid":2,"piid":1,"value":true,"code":0},{"did":"a","siid":2,"piid":2,"value":5,"code":0},{"did":"other","siid":9,"piid":9,"value":1,"code":0}]`},
|
||||||
|
}
|
||||||
|
for _, test := range tests {
|
||||||
|
t.Run(test.name, func(t *testing.T) {
|
||||||
|
device := fixtureDeviceWithResults(t, []string{test.response}, 0)
|
||||||
|
results, err := device.GetMany(context.Background(), []string{"power", "brightness"})
|
||||||
|
if err == nil || results != nil || !strings.Contains(err.Error(), "protocol") {
|
||||||
|
t.Fatalf("GetMany() = %#v, %v, want nil protocol error", results, err)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeviceGetManyValidatesBeforeNetwork(t *testing.T) {
|
||||||
|
device, testServer := fixtureDeviceWithServer(t, nil)
|
||||||
|
tests := [][]string{{"power", "power"}, {"power", "missing"}, {"power", "write-only"}}
|
||||||
|
for _, names := range tests {
|
||||||
|
if results, err := device.GetMany(context.Background(), names); err == nil || results != nil {
|
||||||
|
t.Fatalf("GetMany(%v) = %#v, %v", names, results, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(testServer.requests) != 2 {
|
||||||
|
t.Fatalf("requests = %d, validation reached network", len(testServer.requests))
|
||||||
|
}
|
||||||
|
empty, err := device.GetMany(context.Background(), nil)
|
||||||
|
if err != nil || empty == nil || len(empty) != 0 {
|
||||||
|
t.Fatalf("GetMany(nil) = %#v, %v", empty, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestDeviceGetManyWaitsOnceAfterAllChunks(t *testing.T) {
|
||||||
|
properties, names := batchPropertyFixture(21)
|
||||||
|
responses := make([]string, 2)
|
||||||
|
for chunk := range responses {
|
||||||
|
start := chunk * 20
|
||||||
|
end := min(start+20, len(names))
|
||||||
|
items := make([]PropertyResult, 0, end-start)
|
||||||
|
for index := start; index < end; index++ {
|
||||||
|
property := properties[names[index]]
|
||||||
|
items = append(items, PropertyResult{DID: "a", SIID: property.SIID, PIID: property.PIID, Value: index})
|
||||||
|
}
|
||||||
|
payload, err := json.Marshal(items)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
responses[chunk] = string(payload)
|
||||||
|
}
|
||||||
|
device := fixtureDeviceWithResults(t, responses, 40*time.Millisecond)
|
||||||
|
device.properties = properties
|
||||||
|
|
||||||
|
started := time.Now()
|
||||||
|
if _, err := device.GetMany(context.Background(), names); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
elapsed := time.Since(started)
|
||||||
|
if elapsed < 30*time.Millisecond || elapsed >= 75*time.Millisecond {
|
||||||
|
t.Fatalf("GetMany() delay = %v, want one approximately 40ms wait", elapsed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func batchPropertyFixture(count int) (map[string]PropertySpec, []string) {
|
||||||
|
properties := make(map[string]PropertySpec, count)
|
||||||
|
names := make([]string, count)
|
||||||
|
for index := range count {
|
||||||
|
name := fmt.Sprintf("property-%02d", index)
|
||||||
|
names[index] = name
|
||||||
|
properties[name] = PropertySpec{Name: name, Type: "int", RW: "r", SIID: 10 + index/10, PIID: index%10 + 1}
|
||||||
|
}
|
||||||
|
return properties, names
|
||||||
|
}
|
||||||
|
|
||||||
func TestDeviceMetadataSnapshotsSupportConcurrentReads(t *testing.T) {
|
func TestDeviceMetadataSnapshotsSupportConcurrentReads(t *testing.T) {
|
||||||
device := fixtureDevice(t)
|
device := fixtureDevice(t)
|
||||||
var waitGroup sync.WaitGroup
|
var waitGroup sync.WaitGroup
|
||||||
|
|||||||
@@ -159,6 +159,12 @@ type PropertyResult struct {
|
|||||||
Message string `json:"message,omitempty"`
|
Message string `json:"message,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type DevicePropertyResult struct {
|
||||||
|
Name string
|
||||||
|
Value any
|
||||||
|
Code int
|
||||||
|
}
|
||||||
|
|
||||||
type ActionRequest struct {
|
type ActionRequest struct {
|
||||||
DID string `json:"did"`
|
DID string `json:"did"`
|
||||||
SIID int `json:"siid"`
|
SIID int `json:"siid"`
|
||||||
|
|||||||
Reference in New Issue
Block a user