mqtt-client init

.editorconfig

+# http://editorconfig.org
+root = true
+
+[*]
+indent_style = space
+indent_size = 2
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[*.md]
+trim_trailing_whitespace = false
+
+[Makefile]
+indent_style = tab

.eslintrc.js

+module.exports = {
+    extends: [require.resolve('@umijs/fabric/dist/eslint')],
+    rules: {
+      '@typescript-eslint/prefer-interface': 0,
+      '@typescript-eslint/no-explicit-any': 0,
+      '@typescript-eslint/array-type': ['error', { default: 'array-simple' }],
+      'no-return-assign': 0,
+      semi: ['error', 'always'],
+      'no-confusing-arrow': 0,
+      'no-console': 0,
+      'max-len': ['error', { code: 120, ignoreComments: true, ignoreStrings: true }],
+      // see https://github.com/prettier/prettier/issues/3847
+      'space-before-function-paren': ['error', { anonymous: 'never', named: 'never', asyncArrow: 'always' }],
+      'no-underscore-dangle': 0,
+      'no-plusplus': 0,
+    },
+  };
+  

.fatherrc.js

+export default {
+    target: 'browser',
+    esm: 'babel',
+    cjs: 'babel',
+    runtimeHelpers: true
+  };
+  

.gitignore

+pids
+logs
+node_modules
+npm-debug.log
+coverage/
+run
+dist
+.DS_Store
+.nyc_output
+config.local.js
+.umi
+.umi-production
+.idea/
+.cache
+yarn.lock
+package-lock.json
+.eslintcache
+.history
+.now

.prettierignore

+/test/fixtures
+**/*.gif
+/dist
+/docs
+/es
+/lib
\ No newline at end of file

.prettierrc.js

+const fabric = require('@umijs/fabric');
+
+module.exports = {
+  ...fabric.prettier,
+  printWidth: 120,
+};

package.json

+{
+  "name": "mqtt-client",
+  "version": "1.0.0",
+  "description": "",
+  "main": "./lib/index.js",
+  "module": "./es/index.js",
+  "scripts": {
+    "build": "father-build",
+    "release": "np --no-cleanup --yolo --tag",
+    "prepublishOnly": "yarn lint && yarn build",
+    "lint": "yarn lint:js && yarn lint:prettier",
+    "lint:js": "eslint --cache --ext .js,.jsx,.ts,.tsx --format=pretty ./src",
+    "lint:fix": "yarn lint:js -- --fix",
+    "lint:prettier": "check-prettier lint",
+    "prettier": "prettier -c --write **/*"
+  },
+  "dependencies": {
+    "@babel/runtime": "^7.10.5",
+    "js-base64": "^3.5.2",
+    "kit_utils": "^1.3.6"
+  },
+  "devDependencies": {
+    "@umijs/fabric": "^2.0.7",
+    "check-prettier": "^1.0.3",
+    "cross-env": "^7.0.2",
+    "father-build": "^1.7.0",
+    "husky": "^2.3.0",
+    "jest": "^25.2.2",
+    "lint-staged": "^9.4.2",
+    "np": "^5.0.3",
+    "npm-run-all": "^4.1.5",
+    "prettier": "^1.18.2",
+    "rimraf": "^3.0.0"
+  },
+  "repository": {
+    "type": "git",
+    "url": "https://g.civnet.cn:8443/test/mqtt-client.git"
+  },
+  "bugs": {
+    "url": "https://g.civnet.cn:8443/test/mqtt-client/issues"
+  },
+  "homepage": "https://g.civnet.cn:8443/test/mqtt-client",
+  "files": [
+    "dist",
+    "es",
+    "lib"
+  ],
+  "lint-staged": {
+    "**/*.{js,ts,json,css,md}": [
+      "yarn prettier",
+      "git add"
+    ],
+    "**/*.{js,ts}": [
+      "yarn lint:fix",
+      "git add"
+    ]
+  },
+  "jest": {
+    "moduleFileExtensions": [
+      "js",
+      "ts"
+    ],
+    "testMatch": [
+      "<rootDir>/src/**/__tests__/**/*.test.js"
+    ],
+    "testPathIgnorePatterns": [
+      "/node_modules/",
+      "/fixtures/"
+    ],
+    "coveragePathIgnorePatterns": [
+      "/node_modules/",
+      "/__tests__/",
+      "/dist/"
+    ]
+  },
+  "publishConfig": {
+    "registry": "https://g.civnet.cn:4873"
+  },
+  "keywords": [],
+  "author": "dengxiaofeng",
+  "license": "MIT"
+}

src/Message.js

+import { ERROR } from './consts';
+import {
+  format,
+  parseUTF8,
+  stringToUTF8,
+  UTF8Length,
+} from './utils';
+
+/* eslint-disable */
+class Message {
+    constructor(newPayload) {
+		this.payload;
+		this._destinationName = '';
+		this._qos = 0;
+		this._retained = false;
+		this._duplicate = false;
+		this._stringPayload = null;
+
+        if (typeof newPayload === "string" ||newPayload instanceof ArrayBuffer || (ArrayBuffer.isView(newPayload) && !(newPayload instanceof DataView))) {
+			payload = newPayload;
+		} else {
+			throw (format(ERROR.INVALID_ARGUMENT, [newPayload, "newPayload"]));
+		}
+	}
+	
+	get payloadString() {
+		if (typeof this.payload === 'string') return this.payload
+    	else return parseUTF8(this.payload, 0, this.payload.length)
+	}
+
+	get payloadBytes() {
+		const payload = this.payload
+		if (typeof payload === 'string') {
+			const buffer = new ArrayBuffer(UTF8Length(payload))
+			const byteStream = new Uint8Array(buffer)
+			stringToUTF8(payload, byteStream, 0)
+			return byteStream
+		} else {
+			return payload
+		}
+	}
+
+	get destinationName() {
+		return this._destinationName
+	}
+
+	set destinationName(newDestinationName) {
+		if (typeof newDestinationName === 'string') this._destinationName = newDestinationName
+		else throw new Error(format(ERROR.INVALID_ARGUMENT, [newDestinationName, 'newDestinationName']))
+	}
+
+	get qos() {
+		return this._qos
+	}
+
+	set qos(newQos) {
+		if (newQos === 0 || newQos === 1 || newQos === 2) this._qos = newQos
+    	else throw new Error('Invalid argument:' + newQos)
+	}
+
+	get retained() {
+		return this._retained
+	}
+
+	set retained(newRetained) {
+		if (typeof newRetained === 'boolean') this._retained = newRetained
+		else throw new Error(format(ERROR.INVALID_ARGUMENT, [newRetained, 'newRetained']))
+	}
+
+	get topic() {
+		return this.destinationName
+	}
+
+	set topic(newTopic) {
+		this.destinationName = newTopic
+	}
+
+	get duplicate() {
+		return this._duplicate
+	}
+
+	set duplicate(newDuplicate) {
+		this._duplicate = newDuplicate
+	}
+
+	get stringPayload() {
+		return this._stringPayload
+	}
+
+	set stringPayload(newStringPayload) {
+		this._stringPayload = newStringPayload
+	}
+
+}
+
+export default Message;
\ No newline at end of file

src/Pinger.js

+import {
+  ERROR,
+  MESSAGE_TYPE,
+} from './consts';
+import { format } from './utils';
+import WireMessage from './WireMessage';
+
+/* eslint-disable */
+const Pinger = function(client, keepAliveInterval) {
+    this._client = client;
+    this._keepAliveInterval = keepAliveInterval*1000;
+    this.isReset = false;
+
+    const pingReq = new WireMessage(MESSAGE_TYPE.PINGREQ).encode();
+
+    const doTimeout = function (pinger) {
+        return function () {
+            return doPing.apply(pinger);
+        };
+    };
+
+    /** @ignore */
+    const doPing = function() {
+        if (!this.isReset) {
+            this._client._trace("Pinger.doPing", "Timed out");
+            this._client._disconnected( ERROR.PING_TIMEOUT.code , format(ERROR.PING_TIMEOUT));
+        } else {
+            this.isReset = false;
+            this._client._trace("Pinger.doPing", "send PINGREQ");
+            this._client.socket.send(pingReq);
+            this.timeout = setTimeout(doTimeout(this), this._keepAliveInterval);
+        }
+    };
+
+    this.reset = function() {
+        this.isReset = true;
+        clearTimeout(this.timeout);
+        if (this._keepAliveInterval > 0)
+            this.timeout = setTimeout(doTimeout(this), this._keepAliveInterval);
+    };
+
+    this.cancel = function() {
+        clearTimeout(this.timeout);
+    };
+};
+
+export default Pinger;
\ No newline at end of file

src/WireMessage.js

+import {
+  MESSAGE_TYPE,
+  MqttProtoIdentifierv3,
+  MqttProtoIdentifierv4,
+} from './consts';
+import {
+  encodeMBI,
+  UTF8Length,
+  writeString,
+  writeUint16,
+} from './utils';
+
+/* eslint-disable */
+// interface WireMessageOptions {
+//     messageIdentifier?: number
+//     mqttVersion?: 3 | 4
+//     clientId?: string
+//     willMessage?: MqttMessage
+//     userName?: string
+//     password?: string
+//     topics?: string[]
+//     requestedQos?: (0 | 1 | 2)[]
+//     payloadMessage?: MqttMessage
+//     cleanSession?: boolean
+//     keepAliveInterval?: number
+//     [key: string]: any
+//   }
+
+class WireMessage {
+  constructor(type, options) {
+    this.type = type;
+    for (var name in options) {
+      if (options.hasOwnProperty(name)) {
+        this[name] = options[name];
+      }
+    }
+  }
+
+  encode() {
+    // Compute the first byte of the fixed header
+    var first = (this.type & 0x0f) << 4;
+
+    /*
+     * Now calculate the length of the variable header + payload by adding up the lengths
+     * of all the component parts
+     */
+
+    var remLength = 0;
+    var topicStrLength = [];
+    var destinationNameLength = 0;
+    var willMessagePayloadBytes;
+
+    // if the message contains a messageIdentifier then we need two bytes for that
+    if (this.messageIdentifier !== undefined) remLength += 2;
+
+    switch (this.type) {
+      // If this a Connect then we need to include 12 bytes for its header
+      case MESSAGE_TYPE.CONNECT:
+        switch (this.mqttVersion) {
+          case 3:
+            remLength += MqttProtoIdentifierv3.length + 3;
+            break;
+          case 4:
+            remLength += MqttProtoIdentifierv4.length + 3;
+            break;
+        }
+
+        remLength += UTF8Length(this.clientId) + 2;
+        if (this.willMessage !== undefined) {
+          remLength += UTF8Length(this.willMessage.destinationName) + 2;
+          // Will message is always a string, sent as UTF-8 characters with a preceding length.
+          willMessagePayloadBytes = this.willMessage.payloadBytes;
+          if (!(willMessagePayloadBytes instanceof Uint8Array))
+            willMessagePayloadBytes = new Uint8Array(payloadBytes);
+          remLength += willMessagePayloadBytes.byteLength + 2;
+        }
+        if (this.userName !== undefined)
+          remLength += UTF8Length(this.userName) + 2;
+        if (this.password !== undefined)
+          remLength += UTF8Length(this.password) + 2;
+        break;
+
+      // Subscribe, Unsubscribe can both contain topic strings
+      case MESSAGE_TYPE.SUBSCRIBE:
+        first |= 0x02; // Qos = 1;
+        for (var i = 0; i < this.topics.length; i++) {
+          topicStrLength[i] = UTF8Length(this.topics[i]);
+          remLength += topicStrLength[i] + 2;
+        }
+        remLength += this.requestedQos.length; // 1 byte for each topic's Qos
+        // QoS on Subscribe only
+        break;
+
+      case MESSAGE_TYPE.UNSUBSCRIBE:
+        first |= 0x02; // Qos = 1;
+        for (var i = 0; i < this.topics.length; i++) {
+          topicStrLength[i] = UTF8Length(this.topics[i]);
+          remLength += topicStrLength[i] + 2;
+        }
+        break;
+
+      case MESSAGE_TYPE.PUBREL:
+        first |= 0x02; // Qos = 1;
+        break;
+
+      case MESSAGE_TYPE.PUBLISH:
+        if (this.payloadMessage.duplicate) first |= 0x08;
+        first = first |= this.payloadMessage.qos << 1;
+        if (this.payloadMessage.retained) first |= 0x01;
+        destinationNameLength = UTF8Length(this.payloadMessage.destinationName);
+        remLength += destinationNameLength + 2;
+        var payloadBytes = this.payloadMessage.payloadBytes;
+        remLength += payloadBytes.byteLength;
+        if (payloadBytes instanceof ArrayBuffer)
+          payloadBytes = new Uint8Array(payloadBytes);
+        else if (!(payloadBytes instanceof Uint8Array))
+          payloadBytes = new Uint8Array(payloadBytes.buffer);
+        break;
+
+      case MESSAGE_TYPE.DISCONNECT:
+        break;
+
+      default:
+        break;
+    }
+
+    // Now we can allocate a buffer for the message
+
+    var mbi = encodeMBI(remLength); // Convert the length to MQTT MBI format
+    var pos = mbi.length + 1; // Offset of start of variable header
+    var buffer = new ArrayBuffer(remLength + pos);
+    var byteStream = new Uint8Array(buffer); // view it as a sequence of bytes
+
+    //Write the fixed header into the buffer
+    byteStream[0] = first;
+    byteStream.set(mbi, 1);
+
+    // If this is a PUBLISH then the variable header starts with a topic
+    if (this.type == MESSAGE_TYPE.PUBLISH)
+      pos = writeString(
+        this.payloadMessage.destinationName,
+        destinationNameLength,
+        byteStream,
+        pos
+      );
+    // If this is a CONNECT then the variable header contains the protocol name/version, flags and keepalive time
+    else if (this.type == MESSAGE_TYPE.CONNECT) {
+      switch (this.mqttVersion) {
+        case 3:
+          byteStream.set(MqttProtoIdentifierv3, pos);
+          pos += MqttProtoIdentifierv3.length;
+          break;
+        case 4:
+          byteStream.set(MqttProtoIdentifierv4, pos);
+          pos += MqttProtoIdentifierv4.length;
+          break;
+      }
+      var connectFlags = 0;
+      if (this.cleanSession) connectFlags = 0x02;
+      if (this.willMessage !== undefined) {
+        connectFlags |= 0x04;
+        connectFlags |= this.willMessage.qos << 3;
+        if (this.willMessage.retained) {
+          connectFlags |= 0x20;
+        }
+      }
+      if (this.userName !== undefined) connectFlags |= 0x80;
+      if (this.password !== undefined) connectFlags |= 0x40;
+      byteStream[pos++] = connectFlags;
+      pos = writeUint16(this.keepAliveInterval, byteStream, pos);
+    }
+
+    // Output the messageIdentifier - if there is one
+    if (this.messageIdentifier !== undefined)
+      pos = writeUint16(this.messageIdentifier, byteStream, pos);
+
+    switch (this.type) {
+      case MESSAGE_TYPE.CONNECT:
+        pos = writeString(
+          this.clientId,
+          UTF8Length(this.clientId),
+          byteStream,
+          pos
+        );
+        if (this.willMessage !== undefined) {
+          pos = writeString(
+            this.willMessage.destinationName,
+            UTF8Length(this.willMessage.destinationName),
+            byteStream,
+            pos
+          );
+          pos = writeUint16(
+            willMessagePayloadBytes.byteLength,
+            byteStream,
+            pos
+          );
+          byteStream.set(willMessagePayloadBytes, pos);
+          pos += willMessagePayloadBytes.byteLength;
+        }
+        if (this.userName !== undefined)
+          pos = writeString(
+            this.userName,
+            UTF8Length(this.userName),
+            byteStream,
+            pos
+          );
+        if (this.password !== undefined)
+          pos = writeString(
+            this.password,
+            UTF8Length(this.password),
+            byteStream,
+            pos
+          );
+        break;
+
+      case MESSAGE_TYPE.PUBLISH:
+        // PUBLISH has a text or binary payload, if text do not add a 2 byte length field, just the UTF characters.
+        byteStream.set(payloadBytes, pos);
+
+        break;
+
+      //    	    case MESSAGE_TYPE.PUBREC:
+      //    	    case MESSAGE_TYPE.PUBREL:
+      //    	    case MESSAGE_TYPE.PUBCOMP:
+      //    	    	break;
+
+      case MESSAGE_TYPE.SUBSCRIBE:
+        // SUBSCRIBE has a list of topic strings and request QoS
+        for (var i = 0; i < this.topics.length; i++) {
+          pos = writeString(this.topics[i], topicStrLength[i], byteStream, pos);
+          byteStream[pos++] = this.requestedQos[i];
+        }
+        break;
+
+      case MESSAGE_TYPE.UNSUBSCRIBE:
+        // UNSUBSCRIBE has a list of topic strings
+        for (var i = 0; i < this.topics.length; i++)
+          pos = writeString(this.topics[i], topicStrLength[i], byteStream, pos);
+        break;
+
+      default:
+      // Do nothing.
+    }
+
+    return buffer;
+  }
+}
+
+
+export default WireMessage;
\ No newline at end of file

src/consts.js

+/* eslint-disable */
+export const MESSAGE_TYPE = {
+    CONNECT: 1,
+    CONNACK: 2,
+    PUBLISH: 3,
+    PUBACK: 4,
+    PUBREC: 5,
+    PUBREL: 6,
+    PUBCOMP: 7,
+    SUBSCRIBE: 8,
+    SUBACK: 9,
+    UNSUBSCRIBE: 10,
+    UNSUBACK: 11,
+    PINGREQ: 12,
+    PINGRESP: 13,
+    DISCONNECT: 14
+};
+
+
+	/**
+	 * Unique message type identifiers, with associated
+	 * associated integer values.
+	 * @private
+	 */
+export const ERROR = {
+        OK: {code:0, text:"AMQJSC0000I OK."},
+        CONNECT_TIMEOUT: {code:1, text:"AMQJSC0001E Connect timed out."},
+        SUBSCRIBE_TIMEOUT: {code:2, text:"AMQJS0002E Subscribe timed out."},
+        UNSUBSCRIBE_TIMEOUT: {code:3, text:"AMQJS0003E Unsubscribe timed out."},
+        PING_TIMEOUT: {code:4, text:"AMQJS0004E Ping timed out."},
+        INTERNAL_ERROR: {code:5, text:"AMQJS0005E Internal error. Error Message: {0}, Stack trace: {1}"},
+        CONNACK_RETURNCODE: {code:6, text:"AMQJS0006E Bad Connack return code:{0} {1}."},
+        SOCKET_ERROR: {code:7, text:"AMQJS0007E Socket error:{0}."},
+        SOCKET_CLOSE: {code:8, text:"AMQJS0008I Socket closed."},
+        MALFORMED_UTF: {code:9, text:"AMQJS0009E Malformed UTF data:{0} {1} {2}."},
+        UNSUPPORTED: {code:10, text:"AMQJS0010E {0} is not supported by this browser."},
+        INVALID_STATE: {code:11, text:"AMQJS0011E Invalid state {0}."},
+        INVALID_TYPE: {code:12, text:"AMQJS0012E Invalid type {0} for {1}."},
+        INVALID_ARGUMENT: {code:13, text:"AMQJS0013E Invalid argument {0} for {1}."},
+        UNSUPPORTED_OPERATION: {code:14, text:"AMQJS0014E Unsupported operation."},
+        INVALID_STORED_DATA: {code:15, text:"AMQJS0015E Invalid data in local storage key={0} value={1}."},
+        INVALID_MQTT_MESSAGE_TYPE: {code:16, text:"AMQJS0016E Invalid MQTT message type {0}."},
+        MALFORMED_UNICODE: {code:17, text:"AMQJS0017E Malformed Unicode string:{0} {1}."},
+        BUFFER_FULL: {code:18, text:"AMQJS0018E Message buffer is full, maximum buffer size: {0}."},
+    };
+
+    /** CONNACK RC Meaning. */
+export const CONNACK_RC = {
+        0:"Connection Accepted",
+        1:"Connection Refused: unacceptable protocol version",
+        2:"Connection Refused: identifier rejected",
+        3:"Connection Refused: server unavailable",
+        4:"Connection Refused: bad user name or password",
+        5:"Connection Refused: not authorized"
+};
+
+
+//MQTT protocol and version          6    M    Q    I    s    d    p    3
+export const MqttProtoIdentifierv3 = [0x00,0x06,0x4d,0x51,0x49,0x73,0x64,0x70,0x03];
+//MQTT proto/version for 311         4    M    Q    T    T    4
+export const MqttProtoIdentifierv4 = [0x00,0x04,0x4d,0x51,0x54,0x54,0x04];
\ No newline at end of file

src/index.js

+import Client from './Client';
+import Message from './Message';
+
+/* eslint-disable */
+const MqttClient = {
+    Client: Client,
+    Message: Message
+};
+
+export default MqttClient;
\ No newline at end of file

src/utils.js

+import {
+  ERROR,
+  MESSAGE_TYPE,
+} from './consts';
+import Message from './Message';
+
+/* eslint-disable */
+/**
+	 * Validate an object's parameter names to ensure they
+	 * match a list of expected variables name for this option
+	 * type. Used to ensure option object passed into the API don't
+	 * contain erroneous parameters.
+	 * @param {Object} obj - User options object
+	 * @param {Object} keys - valid keys and types that may exist in obj.
+	 * @throws {Error} Invalid option parameter found.
+	 * @private
+	 */
+    export const validate = function(obj, keys) {
+        for (var key in obj) {
+            if (obj.hasOwnProperty(key)) {
+                if (keys.hasOwnProperty(key)) {
+                    if (typeof obj[key] !== keys[key])
+                        throw new Error(format(ERROR.INVALID_TYPE, [typeof obj[key], key]));
+                } else {
+                    var errorStr = "Unknown property, " + key + ". Valid properties are:";
+                    for (var validKey in keys)
+                        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
+        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];
+    }
+
+    	/**
+	 * 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;
+		}
+
+    
+
+
+
+