Only send message to each client once #63

davedoesdev · 2013-10-18T21:51:16Z

I verified it fixes MQTTAscoltatori too - will send that pull request once this one is merged.

Closes #62.

mcollina · 2013-10-19T06:55:14Z

test/abstract_server.js

+        setTimeout(function ()
+        {
+            client1.disconnect();
+        }, 500);


setTimeout is bad and should be avoided in tests.
Do this instead:

use QoS 1

count the numer of "publish" events.

call disconnect() on "puback", after verifying the number of events.

There will be one publish event. How do you know unexpected duplicates won't be received if you disconnect straight away?

If you use QoS 1, PUBLISH packets will be sent to all subscribers before sending out a PUBACK to the sender. So, if QoS 1 is implemented correctly, all the duplicates will arrive before the PUBACK. If not, we have one more bug.

mcollina · 2013-10-19T07:00:08Z

the mosca.trail[uid] is never deleted. It is a damn memory leak that we cannot afford.

Using a per-server message counter might be easier and safe. Example algorithm:

every message is added a _dedupId, generated by incrementing the counter.
every client memorize the last _dedupId transmitted.

Ideally the counter should be cyclic at Math.pow(2, 32), but I think a clever solution can be found later.

davedoesdev · 2013-10-19T07:19:28Z

mosca.trail[uid] will be deleted when the message is deleted. Does mosca keep the message in memory forever?

In the _dedupId scheme, what if message delivery is intermingled? Doesn't that constrain mosca so that it will always have to send one message before it sends another?

mcollina · 2013-10-19T07:29:57Z

mosca.trail[uid] will be deleted when the message is deleted. Does mosca keep the message in memory forever?

Nope, you are right, sorry. I thought that would be global, as the options are not passed through all the ascoltatori backends. If it's bound to the message, then multiple Mosca sharing the same 'parent' broker will have the same problem. options can be undefined.

In the _dedupId scheme, what if message delivery is intermingled? Doesn't that constrain mosca so that it will always have to send one message before it sends another?

Nope, the counter is incremented only when a message enter the current Mosca instance (in the first forward callback, you can add the _dedupId to the topic itself). From that, all is in-order, e.g. calling emit('aaa', 1); emit('aaa', 2) will emit 1 and then 2.

davedoesdev · 2013-10-19T08:41:00Z

Can't modify topic - can't set property on primitives:

> wup = "foo"
'foo'
> wup._foo = 90
90
> wup
'foo'
> wup._foo
undefined

davedoesdev · 2013-10-19T08:57:57Z

The other thing I thought is it means each ascoltatore has to call all callbacks registered for a topic before it handles any other messages.

mcollina · 2013-10-19T08:58:31Z

Ok, then we need to ensure that in Ascoltatori the options object is always present, even if it's empty, and add the _dedupId there.
Do you think it's better to do that in Ascoltatori or in Mosca?
If we do it here, we might even create an AddDedupIdAscoltatore that wraps the internal ascoltatore.

davedoesdev · 2013-10-19T09:05:23Z

AddDedupIdAscoltatore is a good idea.

mcollina · 2013-10-19T10:27:42Z

Would you like to add that to the PR?

davedoesdev · 2013-10-19T18:04:17Z

Sure. Mosca or Ascoltatori do you think?

davedoesdev · 2013-10-19T19:04:51Z

Think Ascoltatori is tidiest. Adding AddDedupIdAscoltatore as a DecoratorAscoltatore - might be generally useful to projects other than Mosca for recognising duplicates.

mcollina · 2013-10-19T21:43:39Z

Yes. Let ensure there is always an 'option' obj for each message delivered.

Il giorno sabato 19 ottobre 2013, David Halls ha scritto:

Think Ascoltatori is tidiest.

—
Reply to this email directly or view it on GitHubhttps://github.com//pull/63#issuecomment-26656564
.

davedoesdev · 2013-10-20T06:32:44Z

Actually, AddDedupIdAscoltatore isn't right because messages are published from inside ascoltatori themselves, onto TrieAscoltatore. However, we can modify _setPublish to it fills in options as {} if not passed (instead of undefined).

davedoesdev · 2013-10-20T06:50:09Z

So that works but what if the same options object is used for every message? Do we really want to be making a shallow copy (or using Object.create) for every publish or can we assume fresh options for each publish?

…ing a message to prevent same message being sent twice to a connection. Assumes: - options is always an object and never null - fresh options object is minted for each message

davedoesdev · 2013-10-20T07:28:38Z

OK so I pushed what I have. I tested ascoltatori against it.

davedoesdev · 2013-10-20T07:29:28Z

I still need to fix the test

davedoesdev · 2013-10-20T20:04:39Z

Weird - travis build worked on 0.10 but not on 0.8

davedoesdev · 2013-10-20T20:09:36Z

Hmm, repro'd failure on 0.8. Strange...
client1 doesn't get the publish

mcollina · 2013-10-20T21:58:18Z

very weird. I don't exactly now what's going on.. only that supporting 0.8 is really hard.

davedoesdev · 2013-10-21T06:44:29Z

The MQTTConnection gets and emits publish but it doesn't reach the test app. The publish is also received before the other connection gets and emits puback, which is as expected.

davedoesdev · 2013-10-21T18:48:52Z

Interesting- added a log to readable-stream.Writable.emit (intercepted it) and for that test, it thinks the registered function for publish is:

function (packet) {
    that.setUpTimer();
    packet.topic = rewriteTopic(packet.topic);
    that.server.authorizePublish(that, packet.topic, packet.payload, function(err, success) {
      that.handleAuthorizePublish(err, success, packet);
    });
  }

davedoesdev · 2013-10-21T18:54:14Z

I don't really understand that - from the code, it looks like the tests are using a MqttConnection and not a Client.
Clearly Node 0.10 gets that right but for some reason 0.8 doesn't.
I think that might be the server connection which is displaying, in which case it's not getting to the client connection.

davedoesdev · 2013-10-21T20:39:25Z

I think the problem is on node 0.8, mqtt uses readable-stream which is flushing (or rather, not) differently. We don't wait for the stream to flush (neither does mqtt). I don't think we need to alter the code of mosca (or mqtt) since we don't actually care that it's gone. However, the test gets puback straight away and so then disconnects. (The data hasn't yet been flushed).

davedoesdev · 2013-10-21T20:53:07Z

A setTimeout in the test fixes it :)
i.e. wait for the data to be flushed. I can't find a flush method in readable-stream - I don't think we expose the server connection anyway.

mqtt uses readable-stream which doesn't send data straight away. So wait before disconnecting so the data gets chance to go.

davedoesdev · 2013-10-21T21:28:43Z

OK, build now passes.

mcollina · 2013-10-22T07:23:27Z

lib/client.js

@@ -168,6 +169,13 @@ Client.prototype.actualSend = function(packet, retry) {
 Client.prototype._buildForward = function() {
  var that = this;
  this.forward = function(topic, payload, options, subTopic, initialQoS, cb) {
+    if (options._dedupId === undefined) {
+      options._dedupId = that.server.nextDedupId();
+    } else if (options._dedupId === that._lastDedupId) {


This should be a <=, not a ===.

davedoesdev · 2013-10-22T07:24:53Z

@mcollina please review. I couldn't really think of an alternative to setTimeout for the test in order to let the data flush.
Also, the dedupId solution does assume that ascolatori don't defer calls to any of the callbacks on publish (if they do, the messages and dedupIds will become intermingled.

davedoesdev · 2013-10-22T07:26:18Z

Ah just missed your review, fixing...

mcollina · 2013-10-22T07:30:20Z

It's a right assumption. The callback is not deferred, otherwise the performance will just grow too much.

@davedoesdev regarding the puback, I got the problem. How about using two clients? One for publishing and one for subscribing? That should remove the setTimeout.

…messages.

davedoesdev · 2013-10-22T07:36:41Z

:( It does use two clients. Client 1 subscribes, client 2 publishes. The problem is that puback is returned after sending data to client 1's stream. But readable-stream doesn't seem to flush the data right away.

      var called = 0;

      client1.on("publish", function(packet) {
        expect(packet.topic).to.equal("a/b");
        expect(packet.payload).to.equal("some other data");
        if (!packet.dup) {
          expect(called).to.equal(0);
          called++;
        }
      });

      client1.on("suback", function() {
        buildAndConnect(d, function(client2) {

          client2.on("puback", function() {
            setTimeout(function () {
              expect(called).to.equal(1);
              client1.disconnect();
              client2.disconnect();
            }, 1000);
          });

          client2.publish({
            topic: "a/b",
            payload: "some other data",
            messageId: messageId,
            qos: 1
          });
        });
      });

      client1.subscribe({
        subscriptions: subscriptions,
        messageId: messageId
      });

mcollina · 2013-10-22T07:40:18Z

I'm taking it from now and see if I can remove the damn setTimeout.

davedoesdev · 2013-10-22T07:45:55Z

Thanks. I ended up in the readable-stream module source.

davedoesdev · 2013-10-22T07:50:04Z

I guess one way would be to get client1 to send a message back once it's received it. However, this wouldn't give time for any duplicates to be registered.
I pressed the close button by mistake again.
Anyway, over to you...

mcollina · 2013-10-22T07:58:27Z

Ok, merged, it's fixed now, without a setTimeout. The trick was not checking called before disconnecting.
The disconnect method in MQTT.js sends a MQTT disconnect packet, it does not close the connection straight away, but let the server handle it (so it can receive in-flight packets).

mcollina · 2013-10-22T08:00:31Z

Thank again! This was very important (and long to debug).

davedoesdev · 2013-10-22T08:08:50Z

Yes, that works.

Only send message to each client once

6249535

This was referenced Oct 18, 2013

Multiple responses when using wildcards moscajs/ascoltatori#82

Closed

Overlapped topics #62

Closed

mcollina reviewed Oct 19, 2013
View reviewed changes

davedoesdev closed this Oct 19, 2013

davedoesdev reopened this Oct 19, 2013

Add dedupId to options when we forward messages. Check this when send…

6068b7d

…ing a message to prevent same message being sent twice to a connection. Assumes: - options is always an object and never null - fresh options object is minted for each message

davedoesdev added 2 commits October 20, 2013 08:31

Remove crypto require

ab55176

Fix test to use puback

b5b32f3

Fix test fail on Node 0.8

462b383

mqtt uses readable-stream which doesn't send data straight away. So wait before disconnecting so the data gets chance to go.

mcollina reviewed Oct 22, 2013
View reviewed changes

davedoesdev mentioned this pull request Oct 22, 2013

Fix RedisAscoltatore and default options to {} moscajs/ascoltatori#84

Merged

Make sure always use number for dedupId, and don't process any 'old' …

b59252d

…messages.

davedoesdev closed this Oct 22, 2013

davedoesdev reopened this Oct 22, 2013

mcollina merged commit b59252d into moscajs:master Oct 22, 2013

Only send message to each client once #63

Only send message to each client once #63

Conversation

davedoesdev commented Oct 18, 2013

mcollina Oct 19, 2013

Choose a reason for hiding this comment

davedoesdev Oct 19, 2013

Choose a reason for hiding this comment

mcollina Oct 19, 2013

Choose a reason for hiding this comment

mcollina commented Oct 19, 2013

davedoesdev commented Oct 19, 2013

mcollina commented Oct 19, 2013

davedoesdev commented Oct 19, 2013

davedoesdev commented Oct 19, 2013

mcollina commented Oct 19, 2013

davedoesdev commented Oct 19, 2013

mcollina commented Oct 19, 2013

davedoesdev commented Oct 19, 2013

davedoesdev commented Oct 19, 2013

mcollina commented Oct 19, 2013

davedoesdev commented Oct 20, 2013

davedoesdev commented Oct 20, 2013

davedoesdev commented Oct 20, 2013

davedoesdev commented Oct 20, 2013

davedoesdev commented Oct 20, 2013

davedoesdev commented Oct 20, 2013

mcollina commented Oct 20, 2013

davedoesdev commented Oct 21, 2013

davedoesdev commented Oct 21, 2013

davedoesdev commented Oct 21, 2013

davedoesdev commented Oct 21, 2013

davedoesdev commented Oct 21, 2013

davedoesdev commented Oct 21, 2013

mcollina Oct 22, 2013

Choose a reason for hiding this comment

davedoesdev commented Oct 22, 2013

davedoesdev commented Oct 22, 2013

mcollina commented Oct 22, 2013

davedoesdev commented Oct 22, 2013

mcollina commented Oct 22, 2013

davedoesdev commented Oct 22, 2013

davedoesdev commented Oct 22, 2013

mcollina commented Oct 22, 2013

mcollina commented Oct 22, 2013

davedoesdev commented Oct 22, 2013