Commit 314358e1 authored by 邓晓峰's avatar 邓晓峰

fix: WireMessage

parent c1b555d7
......@@ -12,7 +12,7 @@ config.local.js
.umi-production
.idea/
.cache
yarn.lock
yarn.loc
package-lock.json
.eslintcache
.history
......
......@@ -21,6 +21,8 @@ var Timeout = function Timeout(client, timeoutSeconds, action, args) {
};
};
var version = "@VERSION@-@BUILDLEVEL@";
var scope = function scope(f, _scope) {
return function () {
return f.apply(_scope, arguments);
......
......@@ -34,6 +34,8 @@ var Timeout = function Timeout(client, timeoutSeconds, action, args) {
};
};
var version = "@VERSION@-@BUILDLEVEL@";
var scope = function scope(f, _scope) {
return function () {
return f.apply(_scope, arguments);
......
......@@ -26,19 +26,18 @@ import {
// [key: string]: any
// }
class WireMessage {
constructor(type, options) {
const WireMessage = function (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;
WireMessage.prototype.encode = function() {
// 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
......@@ -51,12 +50,13 @@ class WireMessage {
var willMessagePayloadBytes;
// if the message contains a messageIdentifier then we need two bytes for that
if (this.messageIdentifier !== undefined) remLength += 2;
if (this.messageIdentifier !== undefined)
remLength += 2;
switch (this.type) {
switch(this.type) {
// If this a Connect then we need to include 12 bytes for its header
case MESSAGE_TYPE.CONNECT:
switch (this.mqttVersion) {
switch(this.mqttVersion) {
case 3:
remLength += MqttProtoIdentifierv3.length + 3;
break;
......@@ -72,7 +72,7 @@ class WireMessage {
willMessagePayloadBytes = this.willMessage.payloadBytes;
if (!(willMessagePayloadBytes instanceof Uint8Array))
willMessagePayloadBytes = new Uint8Array(payloadBytes);
remLength += willMessagePayloadBytes.byteLength + 2;
remLength += willMessagePayloadBytes.byteLength +2;
}
if (this.userName !== undefined)
remLength += UTF8Length(this.userName) + 2;
......@@ -83,7 +83,7 @@ class WireMessage {
// Subscribe, Unsubscribe can both contain topic strings
case MESSAGE_TYPE.SUBSCRIBE:
first |= 0x02; // Qos = 1;
for (var i = 0; i < this.topics.length; i++) {
for ( var i = 0; i < this.topics.length; i++) {
topicStrLength[i] = UTF8Length(this.topics[i]);
remLength += topicStrLength[i] + 2;
}
......@@ -93,7 +93,7 @@ class WireMessage {
case MESSAGE_TYPE.UNSUBSCRIBE:
first |= 0x02; // Qos = 1;
for (var i = 0; i < this.topics.length; i++) {
for ( var i = 0; i < this.topics.length; i++) {
topicStrLength[i] = UTF8Length(this.topics[i]);
remLength += topicStrLength[i] + 2;
}
......@@ -105,7 +105,7 @@ class WireMessage {
case MESSAGE_TYPE.PUBLISH:
if (this.payloadMessage.duplicate) first |= 0x08;
first = first |= this.payloadMessage.qos << 1;
first = first |= (this.payloadMessage.qos << 1);
if (this.payloadMessage.retained) first |= 0x01;
destinationNameLength = UTF8Length(this.payloadMessage.destinationName);
remLength += destinationNameLength + 2;
......@@ -133,17 +133,13 @@ class WireMessage {
//Write the fixed header into the buffer
byteStream[0] = first;
byteStream.set(mbi, 1);
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
);
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:
......@@ -156,61 +152,41 @@ class WireMessage {
break;
}
var connectFlags = 0;
if (this.cleanSession) connectFlags = 0x02;
if (this.willMessage !== undefined) {
if (this.cleanSession)
connectFlags = 0x02;
if (this.willMessage !== undefined ) {
connectFlags |= 0x04;
connectFlags |= this.willMessage.qos << 3;
connectFlags |= (this.willMessage.qos<<3);
if (this.willMessage.retained) {
connectFlags |= 0x20;
}
}
if (this.userName !== undefined) connectFlags |= 0x80;
if (this.password !== undefined) connectFlags |= 0x40;
if (this.userName !== undefined)
connectFlags |= 0x80;
if (this.password !== undefined)
connectFlags |= 0x40;
byteStream[pos++] = connectFlags;
pos = writeUint16(this.keepAliveInterval, byteStream, pos);
pos = writeUint16 (this.keepAliveInterval, byteStream, pos);
}
// Output the messageIdentifier - if there is one
if (this.messageIdentifier !== undefined)
pos = writeUint16(this.messageIdentifier, byteStream, pos);
pos = writeUint16 (this.messageIdentifier, byteStream, pos);
switch (this.type) {
switch(this.type) {
case MESSAGE_TYPE.CONNECT:
pos = writeString(
this.clientId,
UTF8Length(this.clientId),
byteStream,
pos
);
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
);
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
);
pos = writeString(this.userName, UTF8Length(this.userName), byteStream, pos);
if (this.password !== undefined)
pos = writeString(
this.password,
UTF8Length(this.password),
byteStream,
pos
);
pos = writeString(this.password, UTF8Length(this.password), byteStream, pos);
break;
case MESSAGE_TYPE.PUBLISH:
......@@ -226,7 +202,7 @@ class WireMessage {
case MESSAGE_TYPE.SUBSCRIBE:
// SUBSCRIBE has a list of topic strings and request QoS
for (var i = 0; i < this.topics.length; i++) {
for (var i=0; i<this.topics.length; i++) {
pos = writeString(this.topics[i], topicStrLength[i], byteStream, pos);
byteStream[pos++] = this.requestedQos[i];
}
......@@ -234,7 +210,7 @@ class WireMessage {
case MESSAGE_TYPE.UNSUBSCRIBE:
// UNSUBSCRIBE has a list of topic strings
for (var i = 0; i < this.topics.length; i++)
for (var i=0; i<this.topics.length; i++)
pos = writeString(this.topics[i], topicStrLength[i], byteStream, pos);
break;
......@@ -243,8 +219,6 @@ class WireMessage {
}
return buffer;
}
}
};
export default WireMessage;
\ No newline at end of file
export default WritableStream
\ No newline at end of file
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