fix: remove close listener before next startStreamProcess loop

[patrykwegrzyn]

This cause memory leak during fast publishing when leveldb store is used
VM38 warning.js:27 (node:2090114) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 close listeners added to [MqttClient]. Use emitter.setMaxListeners() to increase limit

[codecov]

Codecov Report

Patch coverage: 100.00% and no project coverage change.

Comparison is base (98e9a46) 93.63% compared to head (f9a8c28) 93.63%.

❗ Current head f9a8c28 differs from pull request most recent head ff9f040. Consider uploading reports for the commit ff9f040 to get more accurate results

Additional details and impacted files

@@           Coverage Diff           @@
##           master    #1038   +/-   ##
=======================================
  Coverage   93.63%   93.63%           
=======================================
  Files           8        8           
  Lines         942      943    +1     
  Branches      249      249           
=======================================
+ Hits          882      883    +1     
  Misses         60       60           

Impacted Files	Coverage Δ
lib/client.js	96.65% <100.00%> (+<0.01%)	⬆️

☔ View full report in Codecov by Sentry.
📢 Do you have feedback about the report comment? Let us know in this issue.

[rtuin]

I can't tell much about if this is implemented correctly, but I know it solves the same problem for me.

[YoDaMa]

@rtuin can you explain the problem in a bit more detail? There is a memory leak when publishing in general or just when using leveldb store?

[rtuin]

@YoDaMa I ran into this issue while writing a simple load generator to run some tests.

• Notice I'm using async-mqtt
• The MQTT broker software is VerneMQ
• The Promise.allConcurrent is from: https://gist.github.com/jcouyang/632709f30e12a7879a73e9e132c0d56b

Let me elaborate with a code sample. Hope this helps!

var mqtt = require('async-mqtt');
const fs = require('fs');

const runAsync = async () => {
    var client = await mqtt.connect('mqtts://my-host.domain', {
        key: fs.readFileSync('client.key'),
        cert: fs.readFileSync('client.crt'),
        ca: [ fs.readFileSync('ca.crt') ]
    });
     
    let count = 0;
    const publishes = [];
    while (count++ < 100000) {
        publishes.push(() => {
            return client.publish('my-topic', 'Hello mqtt' + count, {qos: 2});
        });
    }

    await Promise.allConcurrent(1000)(publishes).then((x) => {
        console.log(x);
    });

    return await client.end();
};

runAsync();

[robertsLando]

@patrykwegrzyn Can you sync with master so I can merge this?

[robertsLando]

If you wish to continue with this PR please remember to change destination branch from master to main 🙏🏼

[robertsLando]

Supersided by #1759