MQEmitter with Kafka

[uwburn]

I'm trying to implement a Kafka backed MQEmitter to use it with aedes.
I'm still experimenting to see what's the best way to go to leverage Kafka scalability, as replicating all MQTT subscriptions over Kafka might not be ideal, since Kafka expects to work with "few" topics with many consumers on them.

That said, i've been looking after mongodb and redis implementations as a starting point and i have a few questions.

In mongo version, when emit is called and a callback is supplied, depending on conditions between id and lastId the callback is either invoked or stored into a waiting object:

var obj = res.ops[0]
var id = obj._id.toString()
var lastId = that._lastId.toString()
if (id > lastId) {
that._waiting[id] = cb
} else {
cb()
}

Then, when the packets are retrieved from the capped collection and processed, the waiting object is looked into, and if a matching callback is found, it invoked on next tick:

var id = obj._id.toString()
if (that._waiting[id]) {
nextTick(that._waiting[id])
delete that._waiting[id]
}

What is the purpose of this? From what i've understood the invocation of the callback is delayed to when the message is extracted from the capped collection, but i'm not sure if it's just that.

In redis version i've seen there's a LRU cache object thats used in the handler function:

function handler (sub, topic, payload) {
var packet = msgpack.decode(payload)
if (!that._cache.get(packet.id)) {
that._emit(packet.msg)
}
that._cache.set(packet.id, true)
}

Again, i'm not getting the point of storing an entry in the cache and emitting again (but with _emit) the packet (I admit i still need to look into MQEmitter code base).

Ciao (from another italian developer) and thanks for your work with MQTT on Node!

[mcollina]

waiting in MongoDB is needed to avoid the read-after-write problem. Basically we are waiting to see the event coming from the oplog before calling the callback.

The LRU in Redis is needed to support deduplication.

[uwburn]

Thanks for the input.

I will try spinning up some docker image to do some experiments. I'll let you know if i come up with something interesting.