Skip to content

M
mqtt-client
Commit c24e19a1 authored by 邓晓峰's avatar 邓晓峰
Browse files
Options

fix: utils decodemessage

parent 130e1683
Hide whitespace changes
Inline Side-by-side
Showing with 270 additions and 276 deletions
es/WireMessage.js
View file @ c24e19a1
...@@ -209,4 +209,4 @@ WireMessage.prototype.encode = function () { ...@@ -209,4 +209,4 @@ WireMessage.prototype.encode = function () {
return buffer; return buffer;
}; };
export default WritableStream; export default WireMessage;
\ No newline at end of file \ No newline at end of file
lib/WireMessage.js
View file @ c24e19a1
...@@ -217,5 +217,5 @@ WireMessage.prototype.encode = function () { ...@@ -217,5 +217,5 @@ WireMessage.prototype.encode = function () {
return buffer; return buffer;
}; };
var _default = WritableStream; var _default = WireMessage;
exports.default = _default; exports.default = _default;
\ No newline at end of file
src/utils.js
View file @ c24e19a1
...@@ -3,290 +3,283 @@ import { ...@@ -3,290 +3,283 @@ import {
MESSAGE_TYPE, MESSAGE_TYPE,
} from './consts'; } from './consts';
import Message from './Message'; import Message from './Message';
import WireMessage from './WireMessage';
/* eslint-disable */ /* eslint-disable */
/** /**
* Validate an object's parameter names to ensure they * Validate an object's parameter names to ensure they
* match a list of expected variables name for this option * match a list of expected variables name for this option
* type. Used to ensure option object passed into the API don't * type. Used to ensure option object passed into the API don't
* contain erroneous parameters. * contain erroneous parameters.
* @param {Object} obj - User options object * @param {Object} obj - User options object
* @param {Object} keys - valid keys and types that may exist in obj. * @param {Object} keys - valid keys and types that may exist in obj.
* @throws {Error} Invalid option parameter found. * @throws {Error} Invalid option parameter found.
* @private * @private
*/ */
export const validate = function(obj, keys) { export const validate = function (obj, keys) {
for (var key in obj) { for (var key in obj) {
if (obj.hasOwnProperty(key)) { if (obj.hasOwnProperty(key)) {
if (keys.hasOwnProperty(key)) { if (keys.hasOwnProperty(key)) {
if (typeof obj[key] !== keys[key]) if (typeof obj[key] !== keys[key]) throw new Error(format(ERROR.INVALID_TYPE, [typeof obj[key], key]));
throw new Error(format(ERROR.INVALID_TYPE, [typeof obj[key], key])); } else {
} else { var errorStr = 'Unknown property, ' + key + '. Valid properties are:';
var errorStr = "Unknown property, " + key + ". Valid properties are:"; for (var validKey in keys) if (keys.hasOwnProperty(validKey)) errorStr = errorStr + ' ' + validKey;
for (var validKey in keys) throw new Error(errorStr);
if (keys.hasOwnProperty(validKey)) }
errorStr = errorStr+" "+validKey;
throw new Error(errorStr);
}
}
}
};
/**
* Format an error message text.
* @private
* @param {error} ERROR value above.
* @param {substitutions} [array] substituted into the text.
* @return the text with the substitutions made.
*/
export const format = function(error, substitutions) {
var text = error.text;
if (substitutions) {
var field,start;
for (var i=0; i<substitutions.length; i++) {
field = "{"+i+"}";
start = text.indexOf(field);
if(start > 0) {
var part1 = text.substring(0,start);
var part2 = text.substring(start+field.length);
text = part1+substitutions[i]+part2;
}
}
}
return text;
};
export function decodeMessage(input,pos) {
var startingPos = pos;
var first = input[pos];
var type = first >> 4;
var messageInfo = first &= 0x0f;
pos += 1;
// Decode the remaining length (MBI format)
var digit;
var remLength = 0;
var multiplier = 1;
do {
if (pos == input.length) {
return [null,startingPos];
}
digit = input[pos++];
remLength += ((digit & 0x7F) * multiplier);
multiplier *= 128;
} while ((digit & 0x80) !== 0);
var endPos = pos+remLength;
if (endPos > input.length) {
return [null,startingPos];
}
var wireMessage = new WireMessage(type);
switch(type) {
case MESSAGE_TYPE.CONNACK:
var connectAcknowledgeFlags = input[pos++];
if (connectAcknowledgeFlags & 0x01)
wireMessage.sessionPresent = true;
wireMessage.returnCode = input[pos++];
break;
case MESSAGE_TYPE.PUBLISH:
var qos = (messageInfo >> 1) & 0x03;
var len = readUint16(input, pos);
pos += 2;
var topicName = parseUTF8(input, pos, len);
pos += len;
// If QoS 1 or 2 there will be a messageIdentifier
if (qos > 0) {
wireMessage.messageIdentifier = readUint16(input, pos);
pos += 2;
}
var message = new Message(input.subarray(pos, endPos));
if ((messageInfo & 0x01) == 0x01)
message.retained = true;
if ((messageInfo & 0x08) == 0x08)
message.duplicate = true;
message.qos = qos;
message.destinationName = topicName;
wireMessage.payloadMessage = message;
break;
case MESSAGE_TYPE.PUBACK:
case MESSAGE_TYPE.PUBREC:
case MESSAGE_TYPE.PUBREL:
case MESSAGE_TYPE.PUBCOMP:
case MESSAGE_TYPE.UNSUBACK:
wireMessage.messageIdentifier = readUint16(input, pos);
break;
case MESSAGE_TYPE.SUBACK:
wireMessage.messageIdentifier = readUint16(input, pos);
pos += 2;
wireMessage.returnCode = input.subarray(pos, endPos);
break;
default:
break;
}
return [wireMessage,endPos];
} }
}
};
export function writeUint16(input, buffer, offset) { /**
buffer[offset++] = input >> 8; //MSB * Format an error message text.
buffer[offset++] = input % 256; //LSB * @private
return offset; * @param {error} ERROR value above.
* @param {substitutions} [array] substituted into the text.
* @return the text with the substitutions made.
*/
export const format = function (error, substitutions) {
var text = error.text;
if (substitutions) {
var field, start;
for (var i = 0; i < substitutions.length; i++) {
field = '{' + i + '}';
start = text.indexOf(field);
if (start > 0) {
var part1 = text.substring(0, start);
var part2 = text.substring(start + field.length);
text = part1 + substitutions[i] + part2;
}
} }
}
export function writeString(input, utf8Length, buffer, offset) { return text;
offset = writeUint16(utf8Length, buffer, offset); };
stringToUTF8(input, buffer, offset);
return offset + utf8Length; export function decodeMessage(input, pos) {
var startingPos = pos;
var first = input[pos];
var type = first >> 4;
var messageInfo = (first &= 0x0f);
pos += 1;
// Decode the remaining length (MBI format)
var digit;
var remLength = 0;
var multiplier = 1;
do {
if (pos == input.length) {
return [null, startingPos];
} }
digit = input[pos++];
remLength += (digit & 0x7f) * multiplier;
multiplier *= 128;
} while ((digit & 0x80) !== 0);
var endPos = pos + remLength;
if (endPos > input.length) {
return [null, startingPos];
}
var wireMessage = new WireMessage(type);
switch (type) {
case MESSAGE_TYPE.CONNACK:
var connectAcknowledgeFlags = input[pos++];
if (connectAcknowledgeFlags & 0x01) wireMessage.sessionPresent = true;
wireMessage.returnCode = input[pos++];
break;
case MESSAGE_TYPE.PUBLISH:
var qos = (messageInfo >> 1) & 0x03;
var len = readUint16(input, pos);
pos += 2;
var topicName = parseUTF8(input, pos, len);
pos += len;
// If QoS 1 or 2 there will be a messageIdentifier
if (qos > 0) {
wireMessage.messageIdentifier = readUint16(input, pos);
pos += 2;
}
var message = new Message(input.subarray(pos, endPos));
if ((messageInfo & 0x01) == 0x01) message.retained = true;
if ((messageInfo & 0x08) == 0x08) message.duplicate = true;
message.qos = qos;
message.destinationName = topicName;
wireMessage.payloadMessage = message;
break;
case MESSAGE_TYPE.PUBACK:
case MESSAGE_TYPE.PUBREC:
case MESSAGE_TYPE.PUBREL:
case MESSAGE_TYPE.PUBCOMP:
case MESSAGE_TYPE.UNSUBACK:
wireMessage.messageIdentifier = readUint16(input, pos);
break;
case MESSAGE_TYPE.SUBACK:
wireMessage.messageIdentifier = readUint16(input, pos);
pos += 2;
wireMessage.returnCode = input.subarray(pos, endPos);
break;
default:
break;
}
return [wireMessage, endPos];
}
export function writeUint16(input, buffer, offset) {
buffer[offset++] = input >> 8; //MSB
buffer[offset++] = input % 256; //LSB
return offset;
}
export function writeString(input, utf8Length, buffer, offset) {
offset = writeUint16(utf8Length, buffer, offset);
stringToUTF8(input, buffer, offset);
return offset + utf8Length;
}
export function readUint16(buffer, offset) {
return 256 * buffer[offset] + buffer[offset + 1];
}
export function readUint16(buffer, offset) { /**
return 256*buffer[offset] + buffer[offset+1]; * Encodes an MQTT Multi-Byte Integer
* @private
*/
export function encodeMBI(number) {
var output = new Array(1);
var numBytes = 0;
do {
var digit = number % 128;
number = number >> 7;
if (number > 0) {
digit |= 0x80;
} }
output[numBytes++] = digit;
} while (number > 0 && numBytes < 4);
/** return output;
* Encodes an MQTT Multi-Byte Integer }
* @private
*/
export function encodeMBI(number) {
var output = new Array(1);
var numBytes = 0;
do {
var digit = number % 128;
number = number >> 7;
if (number > 0) {
digit |= 0x80;
}
output[numBytes++] = digit;
} while ( (number > 0) && (numBytes<4) );
return output;
}
/**
* Takes a String and calculates its length in bytes when encoded in UTF8.
* @private
*/
export function UTF8Length(input) {
var output = 0;
for (var i = 0; i<input.length; i++)
{
var charCode = input.charCodeAt(i);
if (charCode > 0x7FF)
{
// Surrogate pair means its a 4 byte character
if (0xD800 <= charCode && charCode <= 0xDBFF)
{
i++;
output++;
}
output +=3;
}
else if (charCode > 0x7F)
output +=2;
else
output++;
}
return output;
}
/**
* Takes a String and writes it into an array as UTF8 encoded bytes.
* @private
*/
export function stringToUTF8(input, output, start) {
var pos = start;
for (var i = 0; i<input.length; i++) {
var charCode = input.charCodeAt(i);
// Check for a surrogate pair.
if (0xD800 <= charCode && charCode <= 0xDBFF) {
var lowCharCode = input.charCodeAt(++i);
if (isNaN(lowCharCode)) {
throw new Error(format(ERROR.MALFORMED_UNICODE, [charCode, lowCharCode]));
}
charCode = ((charCode - 0xD800)<<10) + (lowCharCode - 0xDC00) + 0x10000;
}
if (charCode <= 0x7F) {
output[pos++] = charCode;
} else if (charCode <= 0x7FF) {
output[pos++] = charCode>>6 & 0x1F | 0xC0;
output[pos++] = charCode & 0x3F | 0x80;
} else if (charCode <= 0xFFFF) {
output[pos++] = charCode>>12 & 0x0F | 0xE0;
output[pos++] = charCode>>6 & 0x3F | 0x80;
output[pos++] = charCode & 0x3F | 0x80;
} else {
output[pos++] = charCode>>18 & 0x07 | 0xF0;
output[pos++] = charCode>>12 & 0x3F | 0x80;
output[pos++] = charCode>>6 & 0x3F | 0x80;
output[pos++] = charCode & 0x3F | 0x80;
}
}
return output;
}
export function parseUTF8(input, offset, length) {
var output = "";
var utf16;
var pos = offset;
while (pos < offset+length)
{
var byte1 = input[pos++];
if (byte1 < 128)
utf16 = byte1;
else
{
var byte2 = input[pos++]-128;
if (byte2 < 0)
throw new Error(format(ERROR.MALFORMED_UTF, [byte1.toString(16), byte2.toString(16),""]));
if (byte1 < 0xE0) // 2 byte character
utf16 = 64*(byte1-0xC0) + byte2;
else
{
var byte3 = input[pos++]-128;
if (byte3 < 0)
throw new Error(format(ERROR.MALFORMED_UTF, [byte1.toString(16), byte2.toString(16), byte3.toString(16)]));
if (byte1 < 0xF0) // 3 byte character
utf16 = 4096*(byte1-0xE0) + 64*byte2 + byte3;
else
{
var byte4 = input[pos++]-128;
if (byte4 < 0)
throw new Error(format(ERROR.MALFORMED_UTF, [byte1.toString(16), byte2.toString(16), byte3.toString(16), byte4.toString(16)]));
if (byte1 < 0xF8) // 4 byte character
utf16 = 262144*(byte1-0xF0) + 4096*byte2 + 64*byte3 + byte4;
else // longer encodings are not supported
throw new Error(format(ERROR.MALFORMED_UTF, [byte1.toString(16), byte2.toString(16), byte3.toString(16), byte4.toString(16)]));
}
}
}
if (utf16 > 0xFFFF) // 4 byte character - express as a surrogate pair
{
utf16 -= 0x10000;
output += String.fromCharCode(0xD800 + (utf16 >> 10)); // lead character
utf16 = 0xDC00 + (utf16 & 0x3FF); // trail character
}
output += String.fromCharCode(utf16);
}
return output;
}
/**
* Takes a String and calculates its length in bytes when encoded in UTF8.
* @private
*/
export function UTF8Length(input) {
var output = 0;
for (var i = 0; i < input.length; i++) {
var charCode = input.charCodeAt(i);
if (charCode > 0x7ff) {
// Surrogate pair means its a 4 byte character
if (0xd800 <= charCode && charCode <= 0xdbff) {
i++;
output++;
}
output += 3;
} else if (charCode > 0x7f) output += 2;
else output++;
}
return output;
}
/**
* Takes a String and writes it into an array as UTF8 encoded bytes.
* @private
*/
export function stringToUTF8(input, output, start) {
var pos = start;
for (var i = 0; i < input.length; i++) {
var charCode = input.charCodeAt(i);
// Check for a surrogate pair.
if (0xd800 <= charCode && charCode <= 0xdbff) {
var lowCharCode = input.charCodeAt(++i);
if (isNaN(lowCharCode)) {
throw new Error(format(ERROR.MALFORMED_UNICODE, [charCode, lowCharCode]));
}
charCode = ((charCode - 0xd800) << 10) + (lowCharCode - 0xdc00) + 0x10000;
}
if (charCode <= 0x7f) {
output[pos++] = charCode;
} else if (charCode <= 0x7ff) {
output[pos++] = ((charCode >> 6) & 0x1f) | 0xc0;
output[pos++] = (charCode & 0x3f) | 0x80;
} else if (charCode <= 0xffff) {
output[pos++] = ((charCode >> 12) & 0x0f) | 0xe0;
output[pos++] = ((charCode >> 6) & 0x3f) | 0x80;
output[pos++] = (charCode & 0x3f) | 0x80;
} else {
output[pos++] = ((charCode >> 18) & 0x07) | 0xf0;
output[pos++] = ((charCode >> 12) & 0x3f) | 0x80;
output[pos++] = ((charCode >> 6) & 0x3f) | 0x80;
output[pos++] = (charCode & 0x3f) | 0x80;
}
}
return output;
}
export function parseUTF8(input, offset, length) {
var output = '';
var utf16;
var pos = offset;
while (pos < offset + length) {
var byte1 = input[pos++];
if (byte1 < 128) utf16 = byte1;
else {
var byte2 = input[pos++] - 128;
if (byte2 < 0) throw new Error(format(ERROR.MALFORMED_UTF, [byte1.toString(16), byte2.toString(16), '']));
if (byte1 < 0xe0)
// 2 byte character
utf16 = 64 * (byte1 - 0xc0) + byte2;
else {
var byte3 = input[pos++] - 128;
if (byte3 < 0)
throw new Error(format(ERROR.MALFORMED_UTF, [byte1.toString(16), byte2.toString(16), byte3.toString(16)]));
if (byte1 < 0xf0)
// 3 byte character
utf16 = 4096 * (byte1 - 0xe0) + 64 * byte2 + byte3;
else {
var byte4 = input[pos++] - 128;
if (byte4 < 0)
throw new Error(
format(ERROR.MALFORMED_UTF, [
byte1.toString(16),
byte2.toString(16),
byte3.toString(16),
byte4.toString(16),
]),
);
if (byte1 < 0xf8)
// 4 byte character
utf16 = 262144 * (byte1 - 0xf0) + 4096 * byte2 + 64 * byte3 + byte4;
// longer encodings are not supported
else
throw new Error(
format(ERROR.MALFORMED_UTF, [
byte1.toString(16),
byte2.toString(16),
byte3.toString(16),
byte4.toString(16),
]),
);
}
}
}
if (utf16 > 0xffff) {
// 4 byte character - express as a surrogate pair
utf16 -= 0x10000;
output += String.fromCharCode(0xd800 + (utf16 >> 10)); // lead character
utf16 = 0xdc00 + (utf16 & 0x3ff); // trail character
}
output += String.fromCharCode(utf16);
}
return output;
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment