diff --git a/index.js b/index.js index 5618d73..e89e0a8 100644 --- a/index.js +++ b/index.js @@ -162,7 +162,7 @@ class TuyaDevice extends EventEmitter { debug('GET Payload:'); debug(payload); - data = await this._send(buffer); + data = await this._send(buffer, commandByte); } // If data read failed with defined error messages or device uses Protocol 3.2 we need to read differently @@ -249,7 +249,7 @@ class TuyaDevice extends EventEmitter { return new Promise((resolve, reject) => { this._expectRefreshResponseForSequenceN = sequenceN; // Send request - this._send(buffer).then(async data => { + this._send(buffer, CommandType.DP_REFRESH).then(async data => { if (data === 'json obj data unvalid') { // Some devices don't respond to DP_QUERY so, for DPS get commands, fall // back to using SEND with null value. This appears to always work as @@ -421,7 +421,7 @@ class TuyaDevice extends EventEmitter { } // Send request - this._send(buffer).catch(error => { + this._send(buffer, commandByte).catch(error => { if (options.shouldWaitForResponse && !resolvedOrRejected) { resolvedOrRejected = true; reject(error); @@ -470,7 +470,7 @@ class TuyaDevice extends EventEmitter { * @param {Buffer} buffer buffer of data * @returns {Promise} returned data for request */ - _send(buffer) { + _send(buffer, commandByte) { const sequenceNo = this._currentSequenceN; // Retry up to 5 times return pRetry(() => { @@ -480,8 +480,12 @@ class TuyaDevice extends EventEmitter { try { this.client.write(buffer); - // Add resolver function - this._resolvers[sequenceNo] = data => resolve(data); + // Add resolver function. Remember the command byte so that a + // response arriving on a different sequence number can still be + // matched (see the fallback at the end of _packetHandler). + const resolver = data => resolve(data); + resolver.commandByte = commandByte; + this._resolvers[sequenceNo] = resolver; } catch (error) { reject(error); } @@ -739,6 +743,18 @@ class TuyaDevice extends EventEmitter { } _packetHandler(packet) { + // Keep our outgoing sequence counter in sync with the device's. + // The device advances its own sequence number for every packet it sends, + // including unsolicited ones (e.g. the STATUS/DP_REFRESH push it emits right + // after a SET). If we don't account for those, our next request reuses a + // sequence number the device has already consumed, so its reply no longer + // matches the resolver we registered and the pending get()/set() hangs until + // the socket is recycled. Re-syncing here makes request/resolver matching + // survive device-initiated packets. + if (packet.sequenceN > this._currentSequenceN) { + this._currentSequenceN = packet.sequenceN; + } + // Protocol 3.4, 3.5 - Response to Msg 0x03 if (packet.commandByte === CommandType.SESS_KEY_NEG_RES) { if (!this.connectPromise) { @@ -820,6 +836,16 @@ class TuyaDevice extends EventEmitter { packet.commandByte === CommandType.CONTROL_NEW ) && packet.payload === false) { debug('Got SET ack.'); + // This request is finished, so settle and drop its resolver — located by + // command byte, because the ack's sequence number can be offset from the + // one we sent (the device numbers its replies with its own counter). + // Left in place, control resolvers accumulate one per set for the life of + // the session and are never matched again. + const controlMatch = this._newestResolverFor(packet.commandByte); + if (controlMatch !== undefined) { + this._resolvers[controlMatch](packet.payload); + delete this._resolvers[controlMatch]; + } return; } @@ -916,14 +942,69 @@ class TuyaDevice extends EventEmitter { return; } + // Query responses are matched by command byte, not by sequence number. The + // device numbers its replies with its OWN counter, which drifts away from + // ours the moment it sends an unsolicited packet (the STATUS push after a + // set) — so a reply can arrive on a different number than the one we + // registered, and a number that collides with a different pending request. + // Matching the newest outstanding query of the same command byte keeps a + // read from hanging, and stops an unrelated packet that merely collides on + // the sequence number from resolving it with the wrong payload. + if ( + (packet.commandByte === CommandType.DP_QUERY || + packet.commandByte === CommandType.DP_QUERY_NEW) && + typeof this._setResolver !== 'function' + ) { + const match = this._newestResolverFor(packet.commandByte); + const colliding = this._resolvers[packet.sequenceN]; + + if ( + match !== undefined && + (colliding === undefined || colliding.commandByte !== packet.commandByte) + ) { + debug('Matched query response to sequence number ' + match + ' by command byte'); + this._resolvers[match](packet.payload); + delete this._resolvers[match]; + this._expectRefreshResponseForSequenceN = undefined; + return; + } + } + // Call data resolver for sequence number if (packet.sequenceN in this._resolvers) { - this._resolvers[packet.sequenceN](packet.payload); + const resolver = this._resolvers[packet.sequenceN]; + const resolverIsQuery = + resolver.commandByte === CommandType.DP_QUERY || + resolver.commandByte === CommandType.DP_QUERY_NEW; - // Remove resolver - delete this._resolvers[packet.sequenceN]; - this._expectRefreshResponseForSequenceN = undefined; + // A query is answered by a packet carrying the same command byte, so a + // packet that only collides on the sequence number must not resolve it. + if (!resolverIsQuery || packet.commandByte === resolver.commandByte) { + resolver(packet.payload); + + // Remove resolver + delete this._resolvers[packet.sequenceN]; + this._expectRefreshResponseForSequenceN = undefined; + } + } + } + + /** + * Sequence number of the newest outstanding request registered for + * `commandByte`, or `undefined` if there is none. Used to match device + * responses that do not come back on the sequence number we sent. + * @private + */ + _newestResolverFor(commandByte) { + let match; + for (const key of Object.keys(this._resolvers)) { + const resolver = this._resolvers[key]; + if (resolver && resolver.commandByte === commandByte) { + const seq = Number(key); + if (match === undefined || seq > match) match = seq; + } } + return match; } /** diff --git a/test/sequence.js b/test/sequence.js new file mode 100644 index 0000000..ae12aa2 --- /dev/null +++ b/test/sequence.js @@ -0,0 +1,64 @@ +import test from 'ava'; + +const TuyAPI = require('..'); + +// A Tuya device advances its own sequence number for every packet it sends, +// including unsolicited ones (e.g. the STATUS/DP_REFRESH push emitted right +// after a SET). TuyAPI must re-sync its outgoing counter from received packets: +// otherwise the next request reuses a sequence number the device has already +// consumed, the device answers with a different number, and the resolver we +// registered is never called — so the pending get()/set() hangs until the socket +// is recycled. + +function makeDevice() { + const device = new TuyAPI({id: '22325186db4a2217dc8e', + key: '4226aa407d5c1e2b', + ip: '127.0.0.1', + version: '3.4'}); + // Pretend a session is already established and swallow socket writes. + device._connected = true; + device.client = {write() {}}; + return device; +} + +test('re-syncs the sequence counter from an unsolicited device packet', t => { + const device = makeDevice(); + device._currentSequenceN = 98; + + // The device pushes a STATUS packet using sequence 100; our counter is behind. + device._packetHandler({commandByte: 8, sequenceN: 100, payload: {dps: {3: 15}}}); + + t.is(device._currentSequenceN, 100); +}); + +test('does not rewind the sequence counter on an older packet', t => { + const device = makeDevice(); + device._currentSequenceN = 100; + + device._packetHandler({commandByte: 8, sequenceN: 90, payload: {dps: {3: 15}}}); + + t.is(device._currentSequenceN, 100); +}); + +test('a pending request still resolves after an unsolicited push', t => { + const device = makeDevice(); + let deviceSequenceN = 200; // the last sequence number the device has used + + // The device pushes an unsolicited packet, advancing its own counter. + deviceSequenceN++; + device._packetHandler({commandByte: 8, sequenceN: deviceSequenceN, payload: {dps: {3: 15}}}); + + // Issue a read the way get()/_send() do: claim the next sequence number and + // register a resolver for it. + const requestSequenceN = ++device._currentSequenceN; + let received; + device._resolvers[requestSequenceN] = payload => { + received = payload; + }; + + // The device answers on the sequence number that follows the one it last used. + deviceSequenceN++; + device._packetHandler({commandByte: 16, sequenceN: deviceSequenceN, payload: {dps: {1: true}}}); + + t.is(received.dps['1'], true); +});