687 lines
16 KiB
JavaScript
687 lines
16 KiB
JavaScript
'use strict'
|
|
|
|
const { test } = require('tap')
|
|
const concat = require('concat-stream')
|
|
const { setup, connect, subscribe } = require('./helper')
|
|
const aedes = require('../')
|
|
|
|
function publish (t, s, packet, done) {
|
|
const msgId = packet.messageId
|
|
|
|
s.inStream.write(packet)
|
|
|
|
s.outStream.once('data', function (packet) {
|
|
t.same(packet, {
|
|
cmd: 'pubrec',
|
|
messageId: msgId,
|
|
length: 2,
|
|
dup: false,
|
|
retain: false,
|
|
qos: 0
|
|
}, 'pubrec must match')
|
|
|
|
s.inStream.write({
|
|
cmd: 'pubrel',
|
|
messageId: msgId
|
|
})
|
|
|
|
s.outStream.once('data', function (packet) {
|
|
t.same(packet, {
|
|
cmd: 'pubcomp',
|
|
messageId: msgId,
|
|
length: 2,
|
|
dup: false,
|
|
retain: false,
|
|
qos: 0
|
|
}, 'pubcomp must match')
|
|
|
|
if (done) {
|
|
done()
|
|
}
|
|
})
|
|
})
|
|
}
|
|
|
|
function receive (t, subscriber, expected, done) {
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.not(packet.messageId, expected.messageId, 'messageId must differ')
|
|
|
|
const msgId = packet.messageId
|
|
delete packet.messageId
|
|
delete expected.messageId
|
|
t.same(packet, expected, 'packet must match')
|
|
|
|
subscriber.inStream.write({
|
|
cmd: 'pubrec',
|
|
messageId: msgId
|
|
})
|
|
|
|
subscriber.outStream.once('data', function (packet) {
|
|
subscriber.inStream.write({
|
|
cmd: 'pubcomp',
|
|
messageId: msgId
|
|
})
|
|
t.same(packet, {
|
|
cmd: 'pubrel',
|
|
messageId: msgId,
|
|
length: 2,
|
|
qos: 1,
|
|
retain: false,
|
|
dup: false
|
|
}, 'pubrel must match')
|
|
|
|
if (done) {
|
|
done()
|
|
}
|
|
})
|
|
})
|
|
}
|
|
|
|
test('publish QoS 2', function (t) {
|
|
t.plan(2)
|
|
|
|
const s = connect(setup())
|
|
t.teardown(s.broker.close.bind(s.broker))
|
|
|
|
const packet = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: 'world',
|
|
qos: 2,
|
|
messageId: 42
|
|
}
|
|
publish(t, s, packet)
|
|
})
|
|
|
|
test('subscribe QoS 2', function (t) {
|
|
t.plan(8)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const publisher = connect(setup(broker))
|
|
const subscriber = connect(setup(broker))
|
|
const toPublish = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
messageId: 42,
|
|
dup: false,
|
|
length: 14,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
publish(t, publisher, toPublish)
|
|
|
|
receive(t, subscriber, toPublish)
|
|
})
|
|
})
|
|
|
|
test('publish QoS 2 throws error on write', function (t) {
|
|
t.plan(1)
|
|
|
|
const s = connect(setup())
|
|
t.teardown(s.broker.close.bind(s.broker))
|
|
|
|
s.broker.on('client', function (client) {
|
|
client.connected = false
|
|
client.connecting = false
|
|
|
|
s.inStream.write({
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: 'world',
|
|
qos: 2,
|
|
messageId: 42
|
|
})
|
|
})
|
|
|
|
s.broker.on('clientError', function (client, err) {
|
|
t.equal(err.message, 'connection closed', 'throws error')
|
|
})
|
|
})
|
|
|
|
test('pubrec handler calls done when outgoingUpdate fails (clean=false)', function (t) {
|
|
t.plan(1)
|
|
|
|
const s = connect(setup(), { clean: false })
|
|
t.teardown(s.broker.close.bind(s.broker))
|
|
|
|
const handle = require('../lib/handlers/pubrec.js')
|
|
|
|
s.broker.persistence.outgoingUpdate = function (client, pubrel, done) {
|
|
done(Error('throws error'))
|
|
}
|
|
|
|
handle(s.client, { messageId: 42 }, function done () {
|
|
t.pass('calls done on error')
|
|
})
|
|
})
|
|
|
|
test('client.publish with clean=true subscribption QoS 2', function (t) {
|
|
t.plan(8)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const toPublish = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
messageId: 42,
|
|
dup: false,
|
|
length: 14,
|
|
retain: false
|
|
}
|
|
let brokerClient = null
|
|
|
|
broker.on('client', function (client) {
|
|
brokerClient = client
|
|
|
|
brokerClient.on('error', function (err) {
|
|
t.error(err)
|
|
})
|
|
})
|
|
|
|
const subscriber = connect(setup(broker), { clean: true })
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
t.pass('subscribed')
|
|
receive(t, subscriber, toPublish)
|
|
brokerClient.publish(toPublish, function (err) {
|
|
t.error(err)
|
|
})
|
|
})
|
|
})
|
|
|
|
test('call published method with client with QoS 2', function (t) {
|
|
t.plan(9)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const publisher = connect(setup(broker))
|
|
const subscriber = connect(setup(broker))
|
|
const toPublish = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
messageId: 42,
|
|
dup: false,
|
|
length: 14,
|
|
retain: false
|
|
}
|
|
|
|
broker.published = function (packet, client, cb) {
|
|
// Client is null for all server publishes
|
|
if (packet.topic.split('/')[0] !== '$SYS') {
|
|
t.ok(client, 'client must be passed to published method')
|
|
cb()
|
|
}
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
publish(t, publisher, toPublish)
|
|
|
|
receive(t, subscriber, toPublish)
|
|
})
|
|
})
|
|
|
|
;[true, false].forEach(function (cleanSession) {
|
|
test(`authorized forward publish packets in QoS 2 [clean=${cleanSession}]`, function (t) {
|
|
t.plan(9)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const opts = { clean: cleanSession }
|
|
const publisher = connect(setup(broker), { clientId: 'my-client-xyz-8' })
|
|
const subscriber = connect(setup(broker), { ...opts, clientId: 'abcde' })
|
|
const forwarded = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
dup: false,
|
|
messageId: undefined,
|
|
clientId: 'my-client-xyz-8'
|
|
}
|
|
const expected = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
length: 14,
|
|
dup: false
|
|
}
|
|
broker.authorizeForward = function (client, packet) {
|
|
forwarded.brokerId = broker.id
|
|
forwarded.brokerCounter = broker.counter
|
|
delete packet.nl
|
|
t.same(packet, forwarded, 'forwarded packet must match')
|
|
return packet
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.not(packet.messageId, 42)
|
|
delete packet.messageId
|
|
t.same(packet, expected, 'packet must match')
|
|
})
|
|
|
|
publish(t, publisher, {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
messageId: 42,
|
|
dup: false
|
|
}, function () {
|
|
const stream = broker.persistence.outgoingStream({ id: 'abcde' })
|
|
stream.pipe(concat(function (list) {
|
|
if (cleanSession) {
|
|
t.equal(list.length, 0, 'should have empty item in queue')
|
|
} else {
|
|
t.equal(list.length, 1, 'should have one item in queue')
|
|
}
|
|
}))
|
|
})
|
|
})
|
|
})
|
|
})
|
|
|
|
;[true, false].forEach(function (cleanSession) {
|
|
test(`unauthorized forward publish packets in QoS 2 [clean=${cleanSession}]`, function (t) {
|
|
t.plan(6)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const opts = { clean: cleanSession }
|
|
|
|
const publisher = connect(setup(broker))
|
|
const subscriber = connect(setup(broker), { ...opts, clientId: 'abcde' })
|
|
|
|
broker.authorizeForward = function (client, packet) {
|
|
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.fail('should not receive any packets')
|
|
})
|
|
|
|
publish(t, publisher, {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
messageId: 42,
|
|
dup: false
|
|
}, function () {
|
|
const stream = broker.persistence.outgoingStream({ id: 'abcde' })
|
|
stream.pipe(concat(function (list) {
|
|
t.equal(list.length, 0, 'should empty in queue')
|
|
}))
|
|
})
|
|
})
|
|
})
|
|
})
|
|
|
|
test('subscribe QoS 0, but publish QoS 2', function (t) {
|
|
t.plan(6)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const publisher = connect(setup(broker))
|
|
const subscriber = connect(setup(broker))
|
|
const expected = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 0,
|
|
dup: false,
|
|
length: 12,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 0, function () {
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.same(packet, expected, 'packet must match')
|
|
})
|
|
|
|
publish(t, publisher, {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
messageId: 42,
|
|
dup: false
|
|
})
|
|
})
|
|
})
|
|
|
|
test('subscribe QoS 1, but publish QoS 2', function (t) {
|
|
t.plan(6)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const publisher = connect(setup(broker))
|
|
const subscriber = connect(setup(broker))
|
|
const expected = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 1,
|
|
dup: false,
|
|
length: 14,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 1, function () {
|
|
subscriber.outStream.once('data', function (packet) {
|
|
delete packet.messageId
|
|
t.same(packet, expected, 'packet must match')
|
|
})
|
|
|
|
publish(t, publisher, {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
messageId: 42,
|
|
dup: false
|
|
})
|
|
})
|
|
})
|
|
|
|
test('restore QoS 2 subscriptions not clean', function (t) {
|
|
t.plan(9)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
let subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' })
|
|
const expected = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
dup: false,
|
|
length: 14,
|
|
messageId: 42,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
subscriber.inStream.end()
|
|
|
|
const publisher = connect(setup(broker))
|
|
|
|
subscriber = connect(setup(broker), { clean: false, clientId: 'abcde' }, function (connect) {
|
|
t.equal(connect.sessionPresent, true, 'session present is set to true')
|
|
publish(t, publisher, expected)
|
|
})
|
|
|
|
receive(t, subscriber, expected)
|
|
})
|
|
})
|
|
|
|
test('resend publish on non-clean reconnect QoS 2', function (t) {
|
|
t.plan(8)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const opts = { clean: false, clientId: 'abcde' }
|
|
let subscriber = connect(setup(broker), opts)
|
|
const expected = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
dup: false,
|
|
length: 14,
|
|
messageId: 42,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
subscriber.inStream.end()
|
|
|
|
const publisher = connect(setup(broker))
|
|
|
|
publish(t, publisher, expected, function () {
|
|
subscriber = connect(setup(broker), opts)
|
|
|
|
receive(t, subscriber, expected)
|
|
})
|
|
})
|
|
})
|
|
|
|
test('resend pubrel on non-clean reconnect QoS 2', function (t) {
|
|
t.plan(9)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const opts = { clean: false, clientId: 'abcde' }
|
|
let subscriber = connect(setup(broker), opts)
|
|
const expected = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
dup: false,
|
|
length: 14,
|
|
messageId: 42,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
subscriber.inStream.end()
|
|
|
|
const publisher = connect(setup(broker))
|
|
|
|
publish(t, publisher, expected, function () {
|
|
subscriber = connect(setup(broker), opts)
|
|
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.not(packet.messageId, expected.messageId, 'messageId must differ')
|
|
|
|
const msgId = packet.messageId
|
|
delete packet.messageId
|
|
delete expected.messageId
|
|
t.same(packet, expected, 'packet must match')
|
|
|
|
subscriber.inStream.write({
|
|
cmd: 'pubrec',
|
|
messageId: msgId
|
|
})
|
|
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.same(packet, {
|
|
cmd: 'pubrel',
|
|
messageId: msgId,
|
|
length: 2,
|
|
qos: 1,
|
|
retain: false,
|
|
dup: false
|
|
}, 'pubrel must match')
|
|
|
|
subscriber.inStream.end()
|
|
|
|
subscriber = connect(setup(broker), opts)
|
|
|
|
subscriber.outStream.once('data', function (packet) {
|
|
t.same(packet, {
|
|
cmd: 'pubrel',
|
|
messageId: msgId,
|
|
length: 2,
|
|
qos: 1,
|
|
retain: false,
|
|
dup: false
|
|
}, 'pubrel must match')
|
|
|
|
subscriber.inStream.write({
|
|
cmd: 'pubcomp',
|
|
messageId: msgId
|
|
})
|
|
})
|
|
})
|
|
})
|
|
})
|
|
})
|
|
})
|
|
|
|
test('publish after disconnection', function (t) {
|
|
t.plan(10)
|
|
|
|
const broker = aedes()
|
|
t.teardown(broker.close.bind(broker))
|
|
|
|
const publisher = connect(setup(broker))
|
|
const subscriber = connect(setup(broker))
|
|
const toPublish = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
messageId: 42,
|
|
dup: false,
|
|
length: 14,
|
|
retain: false
|
|
}
|
|
const toPublish2 = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('worl2'),
|
|
qos: 2,
|
|
messageId: 43,
|
|
dup: false,
|
|
length: 14,
|
|
retain: false
|
|
}
|
|
|
|
subscribe(t, subscriber, 'hello', 2, function () {
|
|
publish(t, publisher, toPublish)
|
|
|
|
receive(t, subscriber, toPublish, function () {
|
|
publish(t, publisher, toPublish2)
|
|
})
|
|
})
|
|
})
|
|
|
|
test('multiple publish and store one', function (t) {
|
|
t.plan(2)
|
|
|
|
const broker = aedes()
|
|
|
|
const sid = {
|
|
id: 'abcde'
|
|
}
|
|
const s = connect(setup(broker), { clientId: sid.id })
|
|
const toPublish = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: Buffer.from('world'),
|
|
qos: 2,
|
|
retain: false,
|
|
dup: false,
|
|
messageId: 42
|
|
}
|
|
|
|
let count = 5
|
|
while (count--) {
|
|
s.inStream.write(toPublish)
|
|
}
|
|
let recvcnt = 0
|
|
s.outStream.on('data', function (packet) {
|
|
if (++recvcnt < 5) return
|
|
broker.close(function () {
|
|
broker.persistence.incomingGetPacket(sid, toPublish, function (err, origPacket) {
|
|
delete origPacket.brokerId
|
|
delete origPacket.brokerCounter
|
|
t.same(origPacket, toPublish, 'packet must match')
|
|
t.error(err)
|
|
})
|
|
})
|
|
})
|
|
})
|
|
|
|
test('packet is written to stream after being stored', function (t) {
|
|
const s = connect(setup())
|
|
|
|
const broker = s.broker
|
|
|
|
t.teardown(broker.close.bind(s.broker))
|
|
|
|
let packetStored = false
|
|
|
|
const fn = broker.persistence.incomingStorePacket.bind(broker.persistence)
|
|
|
|
s.broker.persistence.incomingStorePacket = function (client, packet, done) {
|
|
packetStored = true
|
|
t.pass('packet stored')
|
|
fn(client, packet, done)
|
|
}
|
|
|
|
const packet = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: 'world',
|
|
qos: 2,
|
|
messageId: 42
|
|
}
|
|
|
|
publish(t, s, packet)
|
|
|
|
s.outStream.once('data', function (packet) {
|
|
t.equal(packet.cmd, 'pubrec', 'pubrec received')
|
|
t.equal(packetStored, true, 'after packet store')
|
|
t.end()
|
|
})
|
|
})
|
|
|
|
test('not send pubrec when persistence fails to store packet', function (t) {
|
|
t.plan(2)
|
|
|
|
const s = connect(setup())
|
|
const broker = s.broker
|
|
|
|
t.teardown(broker.close.bind(s.broker))
|
|
|
|
s.broker.persistence.incomingStorePacket = function (client, packet, done) {
|
|
t.pass('packet stored')
|
|
done(new Error('store error'))
|
|
}
|
|
s.broker.on('clientError', function (client, err) {
|
|
t.equal(err.message, 'store error')
|
|
})
|
|
|
|
const packet = {
|
|
cmd: 'publish',
|
|
topic: 'hello',
|
|
payload: 'world',
|
|
qos: 2,
|
|
messageId: 42
|
|
}
|
|
|
|
s.inStream.write(packet)
|
|
s.outStream.once('data', function (packet) {
|
|
t.fail('should not have pubrec')
|
|
})
|
|
})
|