Repository: moscajs/aedes (js)

Issue:
[question] How do I properly wipe a disconnected session?
Hi!

Given that MQTTv5 session expiry is not implemented yet, what is the current best way to wipe a persistent (non-clean) session that's been disconnected for a while? 

E g, if a persistent session has been disconnected for a specific time interval, or the number of queued messages has exceeded a certain number, or something else, I want aedes to completely forget about the client/session, both its subscriptions and its current queue of offline messages.

I'm not sure if this is done in aedes or directly in the persistence layer, but if the latter, let's assume that I'm using `aedes-persistence`, i e, memory only persisence.

Thanks in advance!

Pull request description:
feat: emptyOutgoingQueue
This method allows to empty the client outgoing queue based on 

Fixes #473 


Patch:
diff --git a/aedes.d.ts b/aedes.d.ts
index be8f274..e73241f 100644
--- a/aedes.d.ts
+++ b/aedes.d.ts
@@ -91,6 +91,7 @@ declare namespace aedes {
     ): void
     unsubscribe (topicObjects: Subscriptions | Subscription | Subscription[] | UnsubscribePacket, callback?: (error?: Error) => void): void
     close (callback?: () => void): void
+    emptyOutgoingQueue (callback?: () => void): void
   }
 
   interface Aedes extends EventEmitter {
diff --git a/docs/Client.md b/docs/Client.md
index f0fd3df..183969a 100644
--- a/docs/Client.md
+++ b/docs/Client.md
@@ -10,12 +10,14 @@
   - [client.closed](#clientclosed)
   - [client.id](#clientid)
   - [client.clean](#clientclean)
+  - [client.version](#clientversion)
   - [Event: connected](#event-connected)
   - [Event: error](#event-error)
   - [client.publish (packet, [callback])](#clientpublish-packet-callback)
   - [client.subscribe (subscriptions, [callback])](#clientsubscribe-subscriptions-callback)
   - [client.unsubscribe (unsubscriptions, [callback])](#clientunsubscribe-unsubscriptions-callback)
   - [client.close ([callback])](#clientclose-callback)
+  - [client.emptyOutgoingQueue ([callback])](#clientemptyoutgoingqueue-callback)
 
 ## new Client(aedes, stream, request)
 
@@ -140,6 +142,12 @@ Disconnect client
 
 `callback` will be invoked when client is closed.
 
+## client.emptyOutgoingQueue ([callback])
+
+Clear all outgoing messages (QoS > 1) related to this client from persistence
+
+`callback` will be invoked when the operation ends.
+
 [PUBLISH]: https://github.com/mqttjs/mqtt-packet#publish
 [SUBSCRIBE]: https://github.com/mqttjs/mqtt-packet#subscribe
 [UNSUBSCRIBE]: https://github.com/mqttjs/mqtt-packet#unsubscribe
diff --git a/lib/client.js b/lib/client.js
index 481e6b1..5a1ddee 100644
--- a/lib/client.js
+++ b/lib/client.js
@@ -11,6 +11,8 @@ const QoSPacket = require('./qos-packet')
 const handleSubscribe = require('./handlers/subscribe')
 const handleUnsubscribe = require('./handlers/unsubscribe')
 const handle = require('./handlers')
+const { pipeline } = require('readable-stream')
+const { through } = require('./utils')
 
 module.exports = Client
 
@@ -353,4 +355,19 @@ function dequeue () {
   }
 }
 
+Client.prototype.emptyOutgoingQueue = function (done) {
+  const client = this
+  const persistence = client.broker.persistence
+
+  function filter (packet, enc, next) {
+    persistence.outgoingClearMessageId(client, packet, next)
+  }
+
+  pipeline(
+    persistence.outgoingStream(client),
+    through(filter),
+    done
+  )
+}
+
 function noop () {}
diff --git a/lib/handlers/connect.js b/lib/handlers/connect.js
index ff5bb41..7b16a29 100644
--- a/lib/handlers/connect.js
+++ b/lib/handlers/connect.js
@@ -4,7 +4,7 @@ const retimer = require('retimer')
 const { pipeline } = require('readable-stream')
 const write = require('../write')
 const QoSPacket = require('../qos-packet')
-const through = require('through2')
+const { through } = require('../utils')
 const handleSubscribe = require('./subscribe')
 const shortid = require('shortid')
 
@@ -232,7 +232,7 @@ function emptyQueue (arg, done) {
 
   pipeline(
     outgoing,
-    through.obj(function clearQueue (data, enc, next) {
+    through(function clearQueue (data, enc, next) {
       var packet = new QoSPacket(data, client)
       packet.writeCallback = next
       persistence.outgoingUpdate(client, packet, emptyQueueFilter)
diff --git a/lib/handlers/subscribe.js b/lib/handlers/subscribe.js
index 3131eba..7163fa6 100644
--- a/lib/handlers/subscribe.js
+++ b/lib/handlers/subscribe.js
@@ -2,7 +2,7 @@
 
 const fastfall = require('fastfall')
 const Packet = require('aedes-packet')
-const through = require('through2')
+const { through } = require('../utils')
 const { validateTopic } = require('../utils')
 const write = require('../write')
 
@@ -184,7 +184,7 @@ function completeSubscribe (err) {
 
   const persistence = broker.persistence
   const stream = persistence.createRetainedStreamCombi(topics)
-  stream.pipe(through.obj(function sendRetained (packet, enc, cb) {
+  stream.pipe(through(function sendRetained (packet, enc, cb) {
     packet = new Packet({
       cmd: packet.cmd,
       qos: packet.qos,
diff --git a/lib/utils.js b/lib/utils.js
index 9d3e5b2..1bf1bd0 100644
--- a/lib/utils.js
+++ b/lib/utils.js
@@ -1,5 +1,7 @@
 'use strict'
 
+const { Transform } = require('readable-stream')
+
 function validateTopic (topic, message) {
   const end = topic.length - 1
   const endMinus = end - 1
@@ -26,6 +28,14 @@ function validateTopic (topic, message) {
   }
 }
 
+function through (transform) {
+  return new Transform({
+    objectMode: true,
+    transform
+  })
+}
+
 module.exports = {
-  validateTopic
+  validateTopic,
+  through
 }
diff --git a/package.json b/package.json
index ca67545..9a73213 100644
--- a/package.json
+++ b/package.json
@@ -120,7 +120,6 @@
     "retimer": "^2.0.0",
     "reusify": "^1.0.4",
     "shortid": "^2.2.15",
-    "through2": "^3.0.1",
     "uuid": "^7.0.2"
   }
 }


Test patch:
diff --git a/test/helper.js b/test/helper.js
index 1f45f70..fbf7983 100644
--- a/test/helper.js
+++ b/test/helper.js
@@ -2,7 +2,7 @@
 
 const duplexify = require('duplexify')
 const mqtt = require('mqtt-connection')
-const through = require('through2')
+const { through } = require('../lib/utils')
 const util = require('util')
 const aedes = require('../')
 
@@ -28,7 +28,7 @@ function setup (broker) {
 
 function connect (s, opts, connected) {
   s = Object.create(s)
-  s.outStream = s.outStream.pipe(through.obj(filter))
+  s.outStream = s.outStream.pipe(through(filter))
 
   opts = opts || {}
 
diff --git a/test/qos1.js b/test/qos1.js
index cdedbc8..1eba58a 100644
--- a/test/qos1.js
+++ b/test/qos1.js
@@ -128,6 +128,72 @@ test('publish QoS 1 and check offline queue', function (t) {
   })
 })
 
+test('publish QoS 1 and empty offline queue', function (t) {
+  t.plan(13)
+
+  const broker = aedes()
+  t.tearDown(broker.close.bind(broker))
+
+  const publisher = connect(setup(broker), { clean: false })
+  const subscriberClient = {
+    id: 'abcde'
+  }
+  const subscriber = connect(setup(broker), { clean: false, clientId: subscriberClient.id })
+  var expected = {
+    cmd: 'publish',
+    topic: 'hello',
+    qos: 1,
+    dup: false,
+    retain: false
+  }
+  const expectedAck = {
+    cmd: 'puback',
+    retain: false,
+    qos: 0,
+    dup: false,
+    length: 2,
+    messageId: 10
+  }
+  var sent = {
+    cmd: 'publish',
+    topic: 'hello',
+    payload: 'world',
+    qos: 1,
+    messageId: 10,
+    retain: false,
+    dup: false
+  }
+  var queue = []
+  subscribe(t, subscriber, 'hello', 1, function () {
+    publisher.outStream.on('data', function (packet) {
+      t.deepEqual(packet, expectedAck, 'ack packet must patch')
+    })
+    subscriber.outStream.on('data', function (packet) {
+      queue.push(packet)
+      delete packet.payload
+      delete packet.length
+      t.notEqual(packet.messageId, undefined, 'messageId is assigned a value')
+      t.notEqual(packet.messageId, 10, 'messageId should be unique')
+      expected.messageId = packet.messageId
+      t.deepEqual(packet, expected, 'publish packet must patch')
+      if (queue.length === 2) {
+        setImmediate(() => {
+          broker.clients[subscriberClient.id].emptyOutgoingQueue(function () {
+            for (var i = 0; i < queue.length; i++) {
+              broker.persistence.outgoingClearMessageId(subscriberClient, queue[i], function (_, origPacket) {
+                t.equal(!!origPacket, false, 'Packet has been removed')
+              })
+            }
+          })
+        })
+      }
+    })
+    publisher.inStream.write(sent)
+    sent.payload = 'world2world'
+    publisher.inStream.write(sent)
+  })
+})
+
 test('subscribe QoS 1', function (t) {
   t.plan(5)
 
diff --git a/test/retain.js b/test/retain.js
index cfcf770..134ed9d 100644
--- a/test/retain.js
+++ b/test/retain.js
@@ -1,7 +1,7 @@
 'use strict'
 
 const { test } = require('tap')
-const through = require('through2')
+const { through } = require('../lib/utils')
 const Faketimers = require('@sinonjs/fake-timers')
 const { setup, connect, subscribe, noError } = require('./helper')
 const aedes = require('../')
@@ -302,7 +302,7 @@ test('broker not store zero-byte retained messages', function (t) {
       return
     }
     const stream = s.broker.persistence.createRetainedStream(packet.topic)
-    stream.pipe(through.obj(function sendRetained (packet, enc, cb) {
+    stream.pipe(through(function sendRetained (packet, enc, cb) {
       t.fail('not store zero-byte retained messages')
     }))
   })