Skip to content

Commit 643108d

Browse files
joed74joed74
committed
Added basic MQTT5 support
1 parent be026e0 commit 643108d

File tree

2 files changed

+155
-12
lines changed

2 files changed

+155
-12
lines changed

‎README.md

+6-5
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,17 @@
11
# Eclipse Paho JavaScript client
22

3-
[![Build Status](https://travis-ci.org/eclipse/paho.mqtt.javascript.svg?branch=develop)](https://travis-ci.org/eclipse/paho.mqtt.javascript)
4-
53
The Paho JavaScript Client is an MQTT browser-based client library written in Javascript that uses WebSockets to connect to an MQTT Broker.
64

75
## Project description:
86

97
The Paho project has been created to provide reliable open-source implementations of open and standard messaging protocols aimed at new, existing, and emerging applications for Machine-to-Machine (M2M) and Internet of Things (IoT).
108
Paho reflects the inherent physical and cost constraints of device connectivity. Its objectives include effective levels of decoupling between devices and applications, designed to keep markets open and encourage the rapid growth of scalable Web and Enterprise middleware and applications.
119

10+
**This version has basic support for protocol version 5 (only tested with QOS 0)**
11+
12+
- Connect/Subscribe/Unsubscribe and Send messages to MQTT5 broker with empty properties
13+
- Received messages from MQTT5 broker contain new properties object (till now only userProperties)
14+
1215
## Links
1316

1417
- Project Website: [https://www.eclipse.org/paho](https://www.eclipse.org/paho)
@@ -25,9 +28,7 @@ Paho reflects the inherent physical and cost constraints of device connectivity.
2528

2629
### Downloading
2730

28-
A zip file containing the full and a minified version the Javascript client can be downloaded from the [Paho downloads page](https://projects.eclipse.org/projects/iot.paho/downloads)
29-
30-
Alternatively the Javascript client can be downloaded directly from the projects git repository: [https://raw.githubusercontent.com/eclipse/paho.mqtt.javascript/master/src/paho-mqtt.js](https://raw.githubusercontent.com/eclipse/paho.mqtt.javascript/master/src/paho-mqtt.js).
31+
The Javascript client can be downloaded directly from the projects git repository: [https://raw.githubusercontent.com/joed74/paho.mqtt.javascript/master/src/paho-mqtt.js](https://raw.githubusercontent.com/joed74/paho.mqtt.javascript/master/src/paho-mqtt.js).
3132

3233
Please **do not** link directly to this url from your application.
3334

‎src/paho-mqtt.js

+149-7
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,36 @@ function onMessageArrived(message) {
138138
DISCONNECT: 14
139139
};
140140

141+
var PROPERTY_TYPE = {
142+
PayloadFormatIndicator: 1,
143+
MessageExpiryInterval: 2,
144+
ContentType: 3,
145+
ResponseTopic: 8,
146+
CorrelationData: 9,
147+
SubscriptionIdentifier: 11,
148+
SessionExpiryInterval: 17,
149+
AssignedClientIdentifier: 18,
150+
ServerKeepAlive: 19,
151+
AuthenticationMethod: 21,
152+
AuthenticationData: 22,
153+
RequestProblemInformation: 23,
154+
WillDelayInterval: 24,
155+
RequestResponseInformation: 25,
156+
ResponseInformation: 26,
157+
ServerReference: 28,
158+
ReasonString: 31,
159+
ReceiveMaximum: 33,
160+
TopicAliasMaximum: 34,
161+
TopicAlias: 35,
162+
MaximumQoS: 36,
163+
RetainAvailable: 37,
164+
UserProperty: 38,
165+
MaximumPacketSize: 39,
166+
WildcardSubscriptionAvailable: 40,
167+
SubscriptionIdentifierAvailable: 41,
168+
SharedSubscriptionAvailable: 42
169+
};
170+
141171
// Collection of utility methods used to simplify module code
142172
// and promote the DRY pattern.
143173

@@ -247,6 +277,8 @@ function onMessageArrived(message) {
247277
var MqttProtoIdentifierv3 = [0x00,0x06,0x4d,0x51,0x49,0x73,0x64,0x70,0x03];
248278
//MQTT proto/version for 311 4 M Q T T 4
249279
var MqttProtoIdentifierv4 = [0x00,0x04,0x4d,0x51,0x54,0x54,0x04];
280+
//MQTT proto/version for 5
281+
var MqttProtoIdentifierv5 = [0x00,0x04,0x4d,0x51,0x54,0x54,0x05];
250282

251283
/**
252284
* Construct an MQTT wire protocol message.
@@ -309,6 +341,9 @@ function onMessageArrived(message) {
309341
case 4:
310342
remLength += MqttProtoIdentifierv4.length + 3;
311343
break;
344+
case 5:
345+
remLength += MqttProtoIdentifierv5.length + 4; // property byte
346+
break;
312347
}
313348

314349
remLength += UTF8Length(this.clientId) + 2;
@@ -335,6 +370,7 @@ function onMessageArrived(message) {
335370
}
336371
remLength += this.requestedQos.length; // 1 byte for each topic's Qos
337372
// QoS on Subscribe only
373+
if (this.mqttVersion == 5) remLength += 1; // property byte
338374
break;
339375

340376
case MESSAGE_TYPE.UNSUBSCRIBE:
@@ -343,6 +379,7 @@ function onMessageArrived(message) {
343379
topicStrLength[i] = UTF8Length(this.topics[i]);
344380
remLength += topicStrLength[i] + 2;
345381
}
382+
if (this.mqttVersion == 5) remLength += 1; // property byte
346383
break;
347384

348385
case MESSAGE_TYPE.PUBREL:
@@ -355,6 +392,7 @@ function onMessageArrived(message) {
355392
if (this.payloadMessage.retained) first |= 0x01;
356393
destinationNameLength = UTF8Length(this.payloadMessage.destinationName);
357394
remLength += destinationNameLength + 2;
395+
if (this.mqttVersion == 5) remLength += 1; // property length (0)
358396
var payloadBytes = this.payloadMessage.payloadBytes;
359397
remLength += payloadBytes.byteLength;
360398
if (payloadBytes instanceof ArrayBuffer)
@@ -382,8 +420,12 @@ function onMessageArrived(message) {
382420
byteStream.set(mbi,1);
383421

384422
// If this is a PUBLISH then the variable header starts with a topic
385-
if (this.type == MESSAGE_TYPE.PUBLISH)
423+
if (this.type == MESSAGE_TYPE.PUBLISH) {
386424
pos = writeString(this.payloadMessage.destinationName, destinationNameLength, byteStream, pos);
425+
if (this.mqttVersion == 5) {
426+
byteStream[pos++] = 0; // no properties
427+
}
428+
}
387429
// If this is a CONNECT then the variable header contains the protocol name/version, flags and keepalive time
388430

389431
else if (this.type == MESSAGE_TYPE.CONNECT) {
@@ -396,6 +438,10 @@ function onMessageArrived(message) {
396438
byteStream.set(MqttProtoIdentifierv4, pos);
397439
pos += MqttProtoIdentifierv4.length;
398440
break;
441+
case 5:
442+
byteStream.set(MqttProtoIdentifierv5, pos);
443+
pos += MqttProtoIdentifierv5.length;
444+
break;
399445
}
400446
var connectFlags = 0;
401447
if (this.cleanSession)
@@ -413,6 +459,9 @@ function onMessageArrived(message) {
413459
connectFlags |= 0x40;
414460
byteStream[pos++] = connectFlags;
415461
pos = writeUint16 (this.keepAliveInterval, byteStream, pos);
462+
if (this.mqttVersion == 5) {
463+
byteStream[pos++] = 0; // no properties
464+
}
416465
}
417466

418467
// Output the messageIdentifier - if there is one
@@ -447,6 +496,9 @@ function onMessageArrived(message) {
447496
// break;
448497

449498
case MESSAGE_TYPE.SUBSCRIBE:
499+
if (this.mqttVersion == 5) {
500+
byteStream[pos++] = 0; // no properties
501+
}
450502
// SUBSCRIBE has a list of topic strings and request QoS
451503
for (var i=0; i<this.topics.length; i++) {
452504
pos = writeString(this.topics[i], topicStrLength[i], byteStream, pos);
@@ -455,6 +507,9 @@ function onMessageArrived(message) {
455507
break;
456508

457509
case MESSAGE_TYPE.UNSUBSCRIBE:
510+
if (this.mqttVersion == 5) {
511+
byteStream[pos++] = 0; // no properties
512+
}
458513
// UNSUBSCRIBE has a list of topic strings
459514
for (var i=0; i<this.topics.length; i++)
460515
pos = writeString(this.topics[i], topicStrLength[i], byteStream, pos);
@@ -467,7 +522,71 @@ function onMessageArrived(message) {
467522
return buffer;
468523
};
469524

470-
function decodeMessage(input,pos) {
525+
function parseProperties(input,pos,len) {
526+
var endingPos = pos + len;
527+
var propContainer = {};
528+
propContainer.userProperties = {};
529+
while (pos<endingPos) {
530+
var propId = input[pos++];
531+
switch (propId) {
532+
case PROPERTY_TYPE.PayloadFormatIndicator:
533+
case PROPERTY_TYPE.RequestProblemInformation:
534+
case PROPERTY_TYPE.RequestResponseInformation:
535+
case PROPERTY_TYPE.MaximumQoS:
536+
case PROPERTY_TYPE.RetainAvailable:
537+
case PROPERTY_TYPE.WildcardSubscriptionAvailable:
538+
case PROPERTY_TYPE.SubscriptionIdentifierAvailable:
539+
case PROPERTY_TYPE.SharedSubscriptionAvailable:
540+
// Byte
541+
pos++;
542+
break;
543+
case PROPERTY_TYPE.ServerKeepAlive:
544+
case PROPERTY_TYPE.ReceiveMaximum:
545+
case PROPERTY_TYPE.TopicAliasMaximum:
546+
case PROPERTY_TYPE.TopicAlias:
547+
// Two Byte Integer
548+
pos+=2;
549+
break;
550+
case PROPERTY_TYPE.MessageExpiryInterval:
551+
case PROPERTY_TYPE.SessionExpiryInterval:
552+
case PROPERTY_TYPE.WillDelayInterval:
553+
case PROPERTY_TYPE.MaximumPacketSize:
554+
// Four Byte Integer
555+
pos+=4;
556+
break;
557+
case PROPERTY_TYPE.ContentType:
558+
case PROPERTY_TYPE.ResponseTopic:
559+
case PROPERTY_TYPE.AssignedClientIdentifier:
560+
case PROPERTY_TYPE.AuthenticationMethod:
561+
case PROPERTY_TYPE.ResponseInformation:
562+
case PROPERTY_TYPE.ServerReference:
563+
case PROPERTY_TYPE.ReasonString:
564+
// UTF-8 Encoded String
565+
var strLen = readUint16(input, pos);
566+
pos += 2;
567+
pos += strLen;
568+
break;
569+
case PROPERTY_TYPE.UserProperty:
570+
// UTF-8 String Pair
571+
var keyLen = readUint16(input, pos);
572+
pos += 2;
573+
var key = parseUTF8(input, pos, keyLen);
574+
pos += keyLen;
575+
var valLen = readUint16(input, pos);
576+
pos += 2;
577+
var val = parseUTF8(input, pos, valLen);
578+
pos += valLen;
579+
propContainer.userProperties[key] = val;
580+
break;
581+
default:
582+
pos=endingPos;
583+
break;
584+
}
585+
}
586+
return propContainer;
587+
};
588+
589+
function decodeMessage(input,pos,mqttVersion) {
471590
var startingPos = pos;
472591
var first = input[pos];
473592
var type = first >> 4;
@@ -515,6 +634,12 @@ function onMessageArrived(message) {
515634
wireMessage.messageIdentifier = readUint16(input, pos);
516635
pos += 2;
517636
}
637+
var properties = {};
638+
if (mqttVersion == 5) {
639+
var proplen = input[pos++];
640+
if (proplen>0) properties = parseProperties(input, pos, proplen);
641+
pos += proplen;
642+
}
518643

519644
var message = new Message(input.subarray(pos, endPos));
520645
if ((messageInfo & 0x01) == 0x01)
@@ -523,6 +648,7 @@ function onMessageArrived(message) {
523648
message.duplicate = true;
524649
message.qos = qos;
525650
message.destinationName = topicName;
651+
message.properties = properties;
526652
wireMessage.payloadMessage = message;
527653
break;
528654

@@ -896,6 +1022,7 @@ function onMessageArrived(message) {
8961022
throw new Error(format(ERROR.INVALID_STATE, ["not connected"]));
8971023

8981024
var wireMessage = new WireMessage(MESSAGE_TYPE.SUBSCRIBE);
1025+
wireMessage.mqttVersion = this.connectOptions.mqttVersion;
8991026
wireMessage.topics = filter.constructor === Array ? filter : [filter];
9001027
if (subscribeOptions.qos === undefined)
9011028
subscribeOptions.qos = 0;
@@ -953,6 +1080,7 @@ function onMessageArrived(message) {
9531080

9541081
var wireMessage = new WireMessage(MESSAGE_TYPE.PUBLISH);
9551082
wireMessage.payloadMessage = message;
1083+
wireMessage.mqttVersion = this.connectOptions.mqttVersion;
9561084

9571085
if (this.connected) {
9581086
// Mark qos 1 & 2 message as "ACK required"
@@ -1239,7 +1367,7 @@ function onMessageArrived(message) {
12391367
try {
12401368
var offset = 0;
12411369
while(offset < byteArray.length) {
1242-
var result = decodeMessage(byteArray,offset);
1370+
var result = decodeMessage(byteArray,offset,this.connectOptions.mqttVersion);
12431371
var wireMessage = result[0];
12441372
offset = result[1];
12451373
if (wireMessage !== null) {
@@ -1606,8 +1734,17 @@ function onMessageArrived(message) {
16061734
return;
16071735
}
16081736
} else {
1609-
// Otherwise we never had a connection, so indicate that the connect has failed.
1610-
if (this.connectOptions.mqttVersion === 4 && this.connectOptions.mqttVersionExplicit === false) {
1737+
// Otherwise we never had a connection, so indicate that the connect has failed.
1738+
if (this.connectOptions.mqttVersion === 5 && this.connectOptions.mqttVersionExplicit === false) {
1739+
this._trace("Failed to connect V5, dropping back to V4");
1740+
this.connectOptions.mqttVersion = 4;
1741+
if (this.connectOptions.uris) {
1742+
this.hostIndex = 0;
1743+
this._doConnect(this.connectOptions.uris[0]);
1744+
} else {
1745+
this._doConnect(this.uri);
1746+
}
1747+
} else if (this.connectOptions.mqttVersion === 4 && this.connectOptions.mqttVersionExplicit === false) {
16111748
this._trace("Failed to connect V4, dropping back to V3");
16121749
this.connectOptions.mqttVersion = 3;
16131750
if (this.connectOptions.uris) {
@@ -1951,13 +2088,13 @@ function onMessageArrived(message) {
19512088
if (connectOptions.keepAliveInterval === undefined)
19522089
connectOptions.keepAliveInterval = 60;
19532090

1954-
if (connectOptions.mqttVersion > 4 || connectOptions.mqttVersion < 3) {
2091+
if (connectOptions.mqttVersion > 5 || connectOptions.mqttVersion < 3) {
19552092
throw new Error(format(ERROR.INVALID_ARGUMENT, [connectOptions.mqttVersion, "connectOptions.mqttVersion"]));
19562093
}
19572094

19582095
if (connectOptions.mqttVersion === undefined) {
19592096
connectOptions.mqttVersionExplicit = false;
1960-
connectOptions.mqttVersion = 4;
2097+
connectOptions.mqttVersion = 5;
19612098
} else {
19622099
connectOptions.mqttVersionExplicit = true;
19632100
}
@@ -2380,6 +2517,11 @@ function onMessageArrived(message) {
23802517
enumerable: true,
23812518
get: function() { return duplicate; },
23822519
set: function(newDuplicate) {duplicate=newDuplicate;}
2520+
},
2521+
"properties":{
2522+
enumerable: true,
2523+
get: function() { return properties; },
2524+
set: function(newProperties) {properties=newProperties;}
23832525
}
23842526
});
23852527
};

0 commit comments

Comments
 (0)