Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Improve Request Queue Behavior #5 #817

Merged
merged 5 commits into from
Dec 4, 2023
Merged

Improve Request Queue Behavior #5 #817

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
5207424
Keep requests in queue on send failure until expiration
slugzero Jul 1, 2023
d5414f6
add missing await
slugzero Jul 1, 2023
712449a
always send pending reqeuests in fastpoll, no implicit checkin on gen…
slugzero Jul 9, 2023
30b9458
Merge branch 'Koenkk:master' into request-queue-fixes-5
slugzero Nov 26, 2023
f9114a5
Merge branch 'Koenkk:master' into request-queue-fixes-5
slugzero Dec 4, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion src/controller/controller.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -650,7 +650,10 @@ class Controller extends events.EventEmitter {
}

device.updateLastSeen();
device.implicitCheckin();
//no implicit checkin for genPollCtrl data because it might interfere with the explicit checkin
if (!this.isZclDataPayload(dataPayload, dataType) || !dataPayload.frame.isCluster("genPollCtrl")) {
device.implicitCheckin();
}
device.linkquality = dataPayload.linkquality;

let endpoint = device.getEndpoint(dataPayload.endpoint);
Expand Down
5 changes: 2 additions & 3 deletions src/controller/model/endpoint.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,7 +270,7 @@ class Endpoint extends Entity {

if (this.pendingRequests.size === 0) return;

if (this.sendInProgress) {
if (!fastPolling && this.sendInProgress) {
debug.info(`Request Queue (${this.deviceIeeeAddress}/${this.ID}): sendPendingRequests already in progress`);
return;
}
Expand All @@ -296,12 +296,11 @@ class Endpoint extends Entity {
const result = await request.send();
debug.info(`Request Queue (${this.deviceIeeeAddress}/${this.ID}): send success`);
request.resolve(result);
this.pendingRequests.delete(request);
} catch (error) {
debug.info(`Request Queue (${this.deviceIeeeAddress}/${this.ID}): send failed, expires in ` +
`${(request.expires - now) / 1000} seconds`);
request.reject(error);
}
this.pendingRequests.delete(request);
}
}
this.sendInProgress = false;
Expand Down
76 changes: 38 additions & 38 deletions test/controller.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4587,25 +4587,37 @@ describe('Controller', () => {
jest.advanceTimersByTime(10);
return f;
};
const origSendPendingRequests = endpoint.sendPendingRequests;
endpoint.sendPendingRequests = async (fastpoll) => {
const f = await origSendPendingRequests.call(endpoint, fastpoll);
jest.advanceTimersByTime(10);
return f;
};
endpoint.pendingRequests.add(new Request(async () => {}, [], 100));
mocksendZclFrameToEndpoint.mockClear();
mocksendZclFrameToEndpoint.mockImplementation(async () => {throw new Error('Dogs barking too hard')});
const nextTick = new Promise (process.nextTick);
mocksendZclFrameToEndpoint.mockImplementationOnce(async () => {throw new Error('Cats barking too hard');});
mocksendZclFrameToEndpoint.mockImplementationOnce(async () => {throw new Error('Dogs barking too hard');});
let nextTick = new Promise (process.nextTick);
const result = endpoint.write('genOnOff', {onOff: 1}, {disableResponse: true, sendWhen: 'active'});
await nextTick;
expect(mocksendZclFrameToEndpoint).toHaveBeenCalledTimes(1);
const data = {
wasBroadcast: false,
address: '0x129',
frame: ZclFrame.fromBuffer(Zcl.Utils.getCluster("msOccupancySensing").ID, Buffer.from([24,169,10,0,0,24,1])),
endpoint: 1,
linkquality: 50,
groupID: 1,
}

nextTick = new Promise (process.nextTick);
await mockAdapterEvents['zclData'](data);
await nextTick;
expect(mocksendZclFrameToEndpoint).toHaveBeenCalledTimes(2);
Date.now.mockReturnValue(100000);
let error = null;
try {
await mockAdapterEvents['zclData']({
wasBroadcast: false,
address: '0x129',
frame: ZclFrame.fromBuffer(Zcl.Utils.getCluster("msOccupancySensing").ID, Buffer.from([24,169,10,0,0,24,1])),
endpoint: 1,
linkquality: 50,
groupID: 1,
});

try{
await mockAdapterEvents['zclData'](data);
await result;
} catch (e) {
error = e;
Expand Down Expand Up @@ -4649,8 +4661,9 @@ describe('Controller', () => {
mocksendZclFrameToEndpoint.mockReturnValueOnce( {frame: {Payload: new Array( {"attrId": 3, "attrData": "three", "status": 0})}});
mocksendZclFrameToEndpoint.mockReturnValueOnce( {frame: {Payload: new Array( {"attrId": 4, "attrData": "four", "status": 0})}});
let result1, result2: Promise <any>;
let nextTick = new Promise (process.nextTick);
endpoint.write('genOnOff', {onOff: 0, startUpOnOff: 0}, {disableResponse: true, sendWhen: 'active'});
await new Promise (process.nextTick);
await nextTick;
// Queue content:
// 1. empty
// 2. ZCL write 'genOnOff' {onOff: 0, startUpOnOff: 0}
Expand Down Expand Up @@ -4729,37 +4742,24 @@ describe('Controller', () => {
expect(endpoint.pendingRequests.size).toStrictEqual (7);
expect(mocksendZclFrameToEndpoint).toHaveBeenCalledTimes(8);

try {
// Implicit checkin, there are 5 ZclFrames and 2 other requests left in the queue:
await mockAdapterEvents['zclData']({
wasBroadcast: false,
address: '0x129',
frame: ZclFrame.fromBuffer(Zcl.Utils.getCluster("msOccupancySensing").ID, Buffer.from([24,169,10,0,0,24,1])),
endpoint: 1,
linkquality: 50,
groupID: 1,
});
// Implicit checkin, there are 5 ZclFrames and 2 other requests left in the queue:
await mockAdapterEvents['zclData']({
wasBroadcast: false,
address: '0x129',
frame: ZclFrame.fromBuffer(Zcl.Utils.getCluster("msOccupancySensing").ID, Buffer.from([24,169,10,0,0,24,1])),
endpoint: 1,
linkquality: 50,
groupID: 1,
});

await result6;
} catch (e) {
try {
await result3;
} catch (e) {
try {
await result2;
} catch(e) {
error = e;
}
}
}
expect (result4).resolves.toStrictEqual({"3": "three"});
expect (result5).resolves.toStrictEqual({"3": "three"});
await result4;
expect(result4).resolves.toStrictEqual({"3": "three"});
expect(result5).resolves.toStrictEqual({"3": "three"});
expect(mocksendZclFrameToEndpoint).toHaveBeenCalledTimes(13);
expect(mocksendZclFrameToEndpoint.mock.calls[8][3].Payload).toStrictEqual ([{"attrData": 0, "attrId": 16387, "dataType": 48}]);
expect(mocksendZclFrameToEndpoint.mock.calls[9][3].Payload).toStrictEqual ([{"attrData": 1, "attrId": 0, "dataType": 16}]);
expect(mocksendZclFrameToEndpoint.mock.calls[10][3].Payload).toStrictEqual ([{"attrData": 0, "attrId": 0, "dataType": 16}, {"attrData": 0, "attrId": 16387, "dataType": 48}]);
expect(mocksendZclFrameToEndpoint.mock.calls[11][3].Payload).toStrictEqual ([{"attrData": 1, "attrId": 16387, "dataType": 48}]);
expect(error.message).toStrictEqual(`Write 0x129/1 genOnOff({"onOff":1}, {"sendWhen":"active","timeout":10000,"disableResponse":true,"disableRecovery":false,"disableDefaultResponse":true,"direction":0,"srcEndpoint":null,"reservedBits":0,"manufacturerCode":null,"transactionSequenceNumber":null,"writeUndiv":false}) failed (Write 0x129/1 genOnOff({"onOff":1}, {"sendWhen":"active","timeout":10000,"disableResponse":true,"disableRecovery":false,"disableDefaultResponse":true,"direction":0,"srcEndpoint":null,"reservedBits":0,"manufacturerCode":null,"transactionSequenceNumber":null,"writeUndiv":false}) failed (Dogs barking too hard))`);
});

it('Write with sendWhen active, discard messages after expiration', async () => {
Expand Down