fix: outgoingStream when no value is present for a key

asyncPersistence.js

@@ -111,7 +111,7 @@ async function getRetainedValue (db, topic, hasClusters) {
 async function * createWillStream (db, brokers, maxWills) {
   for (const key of await db.lrange(WILLSKEY, 0, maxWills)) {
     const result = await getDecodedValue(db, WILLSKEY, key)
-    if (!brokers || !brokers[key.split(':')[1]]) {
+    if ((result !== undefined) && (!brokers || !brokers[key.split(':')[1]])) {
       yield result
     }
   }
@@ -511,7 +511,10 @@ class AsyncRedisPersistence {
 
     async function * lrangeResult () {
       for (const key of await db.lrange(clientListKey, 0, maxSessionDelivery)) {
-        yield getDecodedValue(db, clientListKey, key)
+        const decoded = await getDecodedValue(db, clientListKey, key)
+        if (decoded !== undefined) {
+          yield decoded
+        }
       }
     }
 

test/own.js

@@ -229,5 +229,42 @@ async function doTest () {
     t.assert.equal(wills.length, 1, 'should only be one will')
     cleanUpPersistence(t, p)
   })
+
+  test('lrange delivers key with no value', async t => {
+    t.plan(1)
+    await cleanDB()
+    const db = await createDB()
+    const clientId = 'ghostClient'
+    const listKey = 'outgoing:' + encodeURIComponent(clientId)
+    // we just add a key to the list with no associated packet
+    await db.rpush(listKey, 'ghostKey')
+
+    const p = await setUpPersistence(t, 'ghost', { conn: db })
+    const instance = p.instance
+    const packets = []
+    for await (const pkt of instance.outgoingStream({ id: clientId })) {
+      packets.push(pkt)
+    }
+    t.assert.equal(packets.length, 0, 'no packets should be delivered')
+    await db.del(listKey)
+    cleanUpPersistence(t, p)
+  })
+
+  test('createWillStream delivers key with no value', async t => {
+    t.plan(1)
+    await cleanDB()
+    const db = await createDB()
+    await db.rpush('will', 'ghostWillKey')
+
+    const p = await setUpPersistence(t, 'ghostWill', { conn: db })
+    const instance = p.instance
+    const wills = []
+    for await (const will of instance.streamWill()) {
+      wills.push(will)
+    }
+    t.assert.equal(wills.length, 0, 'geen wills gevonden als value ontbreekt')
+    await db.del('will')
+    cleanUpPersistence(t, p)
+  })
 }
 doTest()