mqttjs / mqtt-connection Goto Github PK

I don't see any way to set the session present flag in the README, but it's required by the spec. It seems like all options get passed through to mqtt-packet, which I do see has a sessionPresent field in the type defs, but it would be good for the readme to explicitly state that passing options only documented in mqtt-packet will work.

3.2.2.2 Session Present
Position: bit 0 of the Connect Acknowledge Flags.

If the Server accepts a connection with CleanSession set to 1, the Server MUST set Session Present to 0 in the CONNACK packet in addition to setting a zero return code in the CONNACK packet [MQTT-3.2.2-1].

If the Server accepts a connection with CleanSession set to 0, the value set in Session Present depends on whether the Server already has stored Session state for the supplied client ID. If the Server has stored Session state, it MUST set Session Present to 1 in the CONNACK packet [MQTT-3.2.2-2]. If the Server does not have stored Session state, it MUST set Session Present to 0 in the CONNACK packet. This is in addition to setting a zero return code in the CONNACK packet [MQTT-3.2.2-3].

The Session Present flag enables a Client to establish whether the Client and Server have a consistent view about whether there is already stored Session state.

Once the initial setup of a Session is complete, a Client with stored Session state will expect the Server to maintain its stored Session state. In the event that the value of Session Present received by the Client from the Server is not as expected, the Client can choose whether to proceed with the Session or to disconnect. The Client can discard the Session state on both Client and Server by disconnecting, connecting with Clean Session set to 1 and then disconnecting again.

If a server sends a CONNACK packet containing a non-zero return code it MUST set Session Present to 0 [MQTT-3.2.2-4].

please give me a example

Single mqtt message are received as multiple TCP PDU's on client.

How about make mqtt-connection support the generator. I think it will make development going more smooth.

var net = require('net')
Unresolved variable net error when using this

Hi,
First, thank you for maintaining this module. In my implementation, being able to control the workflow and behavior of the MQTT connection is great.

I have been using MQTT.js for about a year, before the server-side code was outsourced to mqtt-connection.

There have been some memory issues with this library, and I'm pretty sure I nailed them down. Also, the example code is kind of lacking - it would be nice to include the event emitters and so forth.

The memory issue is caused by clients that silently disconnect (without notifying the broker, for whatever reason) and so the "disconnect" event or "close" event is never called. Therefore, they end up clogging the clients[] array and crashing the server.

This is the code I came up with. Works pretty well, however, I'm not sure that I need both calls (stream.destroy() and stream.end()). An official response would be great.

// Load mqtt-connection
var mqttCon = require('mqtt-connection');
var server = new net.Server();

// Create MQTT server
server.on('connection', function (stream) {
    // Get client
    var client = mqttCon(stream);

    // Client connected
    client.on('connect', function (packet) {
        // Connection acknowledgement
        client.connack({ returnCode: 0 });

        // Set client ID
        client.id = packet.clientId;

        // Save client's last seen
        client.last_seen = Math.round(new Date().getTime() / 1000);

        // Save in clients[] array
        self.clients[client.id] = client;
    });

    // Client published
    client.on('publish', function (packet) {
        // Save client's last seen
        client.last_seen = Math.round(new Date().getTime() / 1000);
    });

    // Client pinged
    client.on('pingreq', function (packet) {
        // Save client's last seen
        client.last_seen = Math.round(new Date().getTime() / 1000);

        // Respond
        client.pingresp();
    });

    // Client disconnected
    client.on('disconnect', function (packet) {
        // Remove from clients
        if (self.clients[client.id]) {
            delete self.clients[client.id];
        }

        // End stream
        client.stream.destroy();
        client.stream.end();
    });

    // Client closed connection
    client.on('close', function (err) {
        // Remove from clients
        if (self.clients[client.id]) {
            delete self.clients[client.id];
        }

        // End stream
        client.stream.destroy();
        client.stream.end();
    });

    // Client connection error
    client.on('error', function (err) {
        // Remove from clients
        if (self.clients[client.id]) {
            delete self.clients[client.id];
        }

        // End stream
        client.stream.destroy();
        client.stream.end();
    });
}).listen(self.port);

//---------------------------
// Memory "leak" prevention
//
// Clear out old clients
// (that did not communicate
// for more than 15 minutes,
// connection keep alive is 
// set to every 5 minutes)
//---------------------------

setInterval(function () {
    // Kill connections that have not responded in X minutes
    var cutoff = Math.round(new Date().getTime() / 1000) - (60 * 15);

    // Traverse clients array
    for (var i in self.clients) {
        // Get client
        var client = self.clients[i];

        // Time to kill?
        if (client.last_seen < cutoff) {
            // End stream
            client.stream.destroy();
            client.stream.end();

            // Remove from clients
            if (self.clients[client.id]) {
                delete self.clients[client.id];
            }
        }
    }
}, 10000);

From a client perspective:

Regardless of options passed to initial connection, subsequent calls parse as v4, because that.setOptions(connectPacket) in connection.js sets option to returned packet from connack, which doesn't contain a protocolVersion

mqtt-connection version 3.0.0

in connection.js
this.write(opts)
and parseStream.js
function process(chunk, enc, cb) {

A message cost a 100ms delay , I don't know why. the network is very good and no problem.
how can I debug this ?

node version: 8.11.0
working on windows 7.

The client is successfully subscribed.

The broker receives the publish from client A but doesn't forward it to client B

`server.on('connection', function (stream) {
let conn = mqttConn(stream)

// conn connected
conn.on('connect', function (packet) {
	console.log(conn.options);
  	// acknowledge the connect packet
  	conn.connack({ returnCode: 0 });
	})

conn.on('publish', function (packet) {
	times++;
	packet2send = packet;
	console.log(packet);
	conn.pubrec({ messageId: packet.messageId });
	conn.on('pubrel', function (packet) {
		console.log(packet);
		conn.pubcomp({ messageId: packet.messageId });
		stream.setTimeout(10000)
		stream.on('timeout', function () {
	  		conn.publish({
				retain: false,
				qos: 2,
				dup: false,
				length: 14,
				topic: 'hello',
				payload: 'world',
				messageId: 1
			});
	  		conn.on('pubrec', function (packet) {
				console.log('aa' + packet);
	  			conn.pubrel({ messageId: packet.messageId });
				console.log(packet);
				conn.on('pubcomp', function (packet) {
					console.log('aa' + packet);
					console.log(packet);
				});
			})
		});
	})
})

// conn subscribed
conn.on('subscribe', function (packet) {
	console.log(packet);
// send a suback with messageId and granted QoS level
	conn.suback({ granted: [packet.qos], messageId: messageId })
})

// timeout idle streams after 5 minutes
//stream.setTimeout(1000 * 60 * 5)

// connection error handling
conn.on('close', function (packet) { conn.destroy() 
	console.log('close ' + packet);})
conn.on('error', function (packet) { conn.destroy() 
	console.log('error ' + packet);})
conn.on('disconnect', function (packet) { conn.destroy() 
	console.log('disconnect' + packet);})

})`

I can Sindre pify but since you have seen fit to create a promisified version of mqtt.js can that be done for this repo as well? Blanket promisification of a module has its issues. Thx.

After digging a little bit more on the library, I've closed #24 and opening this one as found what it seems a bug, or at least a miss-documentation.

const client = mqttCon(stream, {
    protocolVersion: 5,
    protocolId: 'MQTT',
});
console.log(client.options);

Outputs undefined.

const client = mqttCon(stream);
client.setOptions({
    protocolVersion: 5,
    protocolId: 'MQTT',
});
console.log(client.options);

Outputs { protocolVersion: 5, protocolId: 'MQTT' }.

This has a big implication, as it causes errors on subscriptions (cannot subscribe as topic cannot be decoded) and also on publish events receives payloads with extra 00 byte at the begining, corresponding to undecoded properties:

Server is sending the content with the space for properties but client is not consuming them, so there are some bytes not being consumed and being left in payload.

Also on the contrary, the client sends a packet with properties bytes codified, and then server is not consuming the properties so they remain in payload, making payload to be displaced so cannot be correctly decoded:

from parser.js:

  // Properties mqtt 5
  if (this.settings.protocolVersion === 5) {
    var properties = this._parseProperties()
    if (Object.getOwnPropertyNames(properties).length) {
      packet.properties = properties
    }
  }

I'm trying to use MQTT 5.0 but I'm always getting the same error: Cannot parse topic

        client.connect({
            clientId: activeInterface.mac_address.replace(/:/ig, '').toUpperCase(),
            username: 'bla',
            password: 'bla',
            protocolId: 'MQTT',
            protocolVersion: 5,
        });

        client.on('connack', () => {
            console.log('client connected... sending subscribe');
            client.subscribe({
                messageId: 0,
                subscriptions: [{
                    topic: 'whatever',
                    qos: 0,
                    nl: false, // no Local MQTT 5.0 flag
                    rap: true, // Retain as Published MQTT 5.0 flag
                    rh: 1, // Retain Handling MQTT 5.0
                }],
            });
        });

I've add this line in parser.js:466 for debugging:

console.log(length, end, this._list.length, this.packet.length, this._list.toString('utf8'));

It always fails (whatever topic I place):

2167 2172 13 13 '\u0000\u0000\u0000\bwhatever\u0000'
client error Error: Cannot parse topic
    at Parser._parseSubscribe (/node_modules/mqtt-connection/node_modules/mqtt-packet/parser.js:304:48)
    at Parser._parsePayload (/node_modules/mqtt-connection/node_modules/mqtt-packet/parser.js:99:14)
    at Parser.parse (/node_modules/mqtt-connection/node_modules/mqtt-packet/parser.js:39:41)
    at DestroyableTransform.process [as _transform] (/node_modules/mqtt-connection/lib/parseStream.js:14:17)
    at DestroyableTransform.Transform._read (/node_modules/through2/node_modules/readable-stream/lib/_stream_transform.js:184:10)
    at DestroyableTransform.Transform._write (/node_modules/through2/node_modules/readable-stream/lib/_stream_transform.js:172:83)
    at doWrite (/node_modules/through2/node_modules/readable-stream/lib/_stream_writable.js:428:64)
    at writeOrBuffer (/node_modules/through2/node_modules/readable-stream/lib/_stream_writable.js:417:5)
    at DestroyableTransform.Writable.write (/node_modules/through2/node_modules/readable-stream/lib/_stream_writable.js:334:11)
    at Socket.ondata (_stream_readable.js:639:20)

If I switch to MQTT lower version it works perfectly...

index.d.ts

import { Duplexify } from 'duplexify';
import { Stream, Duplex } from 'stream'

export = Connection;

type ConnectionCallback = ()=>void

interface ConnectionConstructor {
    (duplex?: Duplex, opts?: any | ConnectionCallback, cb?: ConnectionCallback): Connection.Connection;
    new(duplex?: Duplex, opts?: any | ConnectionCallback, cb?: ConnectionCallback): Connection.Connection;
}
declare var Connection: ConnectionConstructor;
declare namespace Connection {
    interface Connection extends Duplexify {
        connect(opts:any, cb?:ConnectionCallback):void;
        connack(opts:any, cb?:ConnectionCallback):void;
        publish(opts:any, cb?:ConnectionCallback):void;
        puback(opts:any, cb?:ConnectionCallback):void;
        pubrec(opts:any, cb?:ConnectionCallback):void;
        pubrel(opts:any, cb?:ConnectionCallback):void;
        pubcomp(opts:any, cb?:ConnectionCallback):void;
        subscribe(opts:any, cb?:ConnectionCallback):void;
        suback(opts:any, cb?:ConnectionCallback):void;
        unsubscribe(opts:any, cb?:ConnectionCallback):void;
        unsuback(opts:any, cb?:ConnectionCallback):void;
        pingreq(opts:any, cb?:ConnectionCallback):void;
        pingresp(opts:any, cb?:ConnectionCallback):void;
        disconnect(opts:any, cb?:ConnectionCallback):void;
        auth(opts:any, cb?:ConnectionCallback):void;

        destroy():void;
        setOptions(opts:any):void;
    }
}

Do documents.