Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

chore: update deps cached-persistence 9.0.0, tape 5.5.3, ioredis 5.0.5 #96

Merged
merged 6 commits into from
May 20, 2022
Merged

chore: update deps cached-persistence 9.0.0, tape 5.5.3, ioredis 5.0.5 #96

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
415c2d0
Update deps cached-persistence 9.0.0
seriousme May 17, 2022
050b4c6
Add support for RH, RAP and NL
seriousme May 18, 2022
ae13de6
Fix messageId check
seriousme May 18, 2022
7d517a7
make subs generic and use msgpack instead of JSON
seriousme May 19, 2022
0600085
Update key handling
seriousme May 19, 2022
82cd439
Add UPGRADE.md
seriousme May 19, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions package.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,12 +58,12 @@
"release-it": "^15.0.0",
"snazzy": "^9.0.0",
"standard": "^17.0.0",
"tape": "^4.15.1"
"tape": "^5.5.3"
},
"dependencies": {
"aedes-cached-persistence": "^8.1.1",
"aedes-cached-persistence": "^9.0.0",
"hashlru": "^2.3.0",
"ioredis": "^5.0.4",
"ioredis": "^5.0.5",
"msgpack-lite": "^0.1.26",
"pump": "^3.0.0",
"qlobber": "^7.0.0",
Expand Down
25 changes: 16 additions & 9 deletions persistence.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ const pump = require('pump')
const CachedPersistence = require('aedes-cached-persistence')
const Packet = CachedPersistence.Packet
const HLRU = require('hashlru')
const QlobberTrue = require('qlobber').QlobberTrue
const { QlobberTrue } = require('qlobber')
const qlobberOpts = {
separator: '/',
wildcard_one: '+',
Expand Down Expand Up @@ -90,7 +90,7 @@ class RedisPersistence extends CachedPersistence {
let errored

for (const sub of subs) {
toStore[sub.topic] = sub.qos
toStore[sub.topic] = JSON.stringify({ qos: sub.qos, rh: sub.rh, rap: sub.rap, nl: sub.nl })
}

this._db.sadd(clientsKey, client.id, finish)
Expand Down Expand Up @@ -267,7 +267,7 @@ class RedisPersistence extends CachedPersistence {

outgoingUpdate (client, packet, cb) {
const that = this
if (packet.brokerId && packet.messageId) {
if ('brokerId' in packet && 'messageId' in packet) {
updateWithClientData(this, client, packet, cb)
} else {
augmentWithBrokerData(this, client, packet, function updateClient (err) {
Expand All @@ -280,8 +280,8 @@ class RedisPersistence extends CachedPersistence {

outgoingClearMessageId (client, packet, cb) {
const that = this
const clientListKey = outgoingKey + client.id
const messageIdKey = `${outgoingIdKey + client.id}:${packet.messageId}`
const clientListKey = `${outgoingKey}${client.id}`
const messageIdKey = `${outgoingIdKey}${client.id}:${packet.messageId}`

const clientKey = this.messageIdCache.get(messageIdKey)
this.messageIdCache.remove(messageIdKey)
Expand Down Expand Up @@ -525,9 +525,13 @@ function returnSubsForClient (subs) {
}

for (const subKey of subKeys) {
const sub = JSON.parse(subs[subKey])
seriousme marked this conversation as resolved.
Show resolved Hide resolved
toReturn.push({
topic: subKey,
qos: parseInt(subs[subKey])
qos: sub.qos,
rh: sub.rh,
rap: sub.rap,
nl: sub.nl
seriousme marked this conversation as resolved.
Show resolved Hide resolved
})
}

Expand All @@ -536,12 +540,15 @@ function returnSubsForClient (subs) {

function processKeysForClient (clientId, clientHash, that) {
const topics = Object.keys(clientHash)

for (const topic of topics) {
const sub = clientHash[topic]
that._trie.add(topic, {
clientId,
topic,
qos: clientHash[topic]
qos: sub.qos,
rh: sub.rh,
rap: sub.rap,
nl: sub.nl
seriousme marked this conversation as resolved.
Show resolved Hide resolved
})
}
}
Expand Down Expand Up @@ -617,7 +624,7 @@ function updateWithClientData (that, client, packet, cb) {
}

function augmentWithBrokerData (that, client, packet, cb) {
const messageIdKey = `${outgoingIdKey + client.id}:${packet.messageId}`
const messageIdKey = `${outgoingIdKey}${client.id}:${packet.messageId}`

const key = that.messageIdCache.get(messageIdKey)
if (!key) {
Expand Down
15 changes: 11 additions & 4 deletions test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,7 @@ test('outgoingUpdate doesn\'t clear packet ttl', t => {

test('multiple persistences', t => {
t.plan(7)
t.timeoutAfter(60 * 1000)
db.flushall()
const emitter = mqemitterRedis()
const emitter2 = mqemitterRedis()
Expand Down Expand Up @@ -186,18 +187,24 @@ test('multiple persistences', t => {
}
}

instance2._waitFor(client, 'sub_' + 'hello', () => {
instance2._waitFor(client, true, 'hello', () => {
instance2.subscriptionsByTopic('hello', (err, resubs) => {
t.notOk(err, 'subs by topic no error')
t.deepEqual(resubs, [{
clientId: client.id,
topic: 'hello/#',
qos: 1
qos: 1,
rh: undefined,
rap: undefined,
nl: undefined
}, {
clientId: client.id,
topic: 'hello',
qos: 1
}])
qos: 1,
rh: undefined,
rap: undefined,
nl: undefined
}], 'received correct subs')
gotSubs = true
close()
})
Expand Down