Hi,
I have tried the version 1.0 of the library, but the problem remains.
I seems that the issue is related to our custom persistence implementation using SQLite on android.
When using the in memory persistence class from the library the problem doesn't occur.
The code below includes some comments added by me explaining the issue when using our custom persistence class and also what happens when the in memory persistance is used.
=========================
MqttWireMessage message = in .readMqttWireMessage(); // A duplicate PUBCOMP message is received while the the same message( with the same msg id) received lets say 20ms ago is in the process of being removed from the persistence.
token = tokenStore.getToken(message); /* CASE 1: In the case of inmemory persistence, the removal of the previous message has finished before this line is executed, so the token is null here. */
/* CASE 2: It seems that our sqlite persistence takes some time for removing the message, which results in token not being null here because it was not removed yet.*/
if (token != null) { // CASE 2: this check is true
synchronized(token) { // The CommsReceiver thread wait's for other threads to finish work with the token.
// I've added this line here for debugging purposes, and here the token is null in CASE 2 (i.e. it has been removed from the persistence and from the TokenStore)
MqttToken tok = tokenStore.getToken(message);
// Ensure the notify processing is done under a lock on the token
// This ensures that the send processing can complete before the
// receive processing starts! ( request and ack and ack processing
// can occur before request processing is complete if not!
clientState.notifyReceivedAck((MqttAck) message);
}
} else {
// CASE 1: The token is null
// It its an ack and there is no token then something is not right.
// An ack should always have a token assoicated with it.
throw new MqttException(MqttException.REASON_CODE_UNEXPECTED_ERROR);
}
================================
A quick fix for this might be checking if the token is null in the synchronized block and throwing an exception if it's null.
Here is a log from the mosquito broker. Here you can see that the broker is sending the same PUBREC (Mid:88) message 3 times and then sending PUBCOMP (Mid:88) 3 times:
----------------------------------------
1414749583: Sending CONNACK to subscriber_id (0)
1414749583: Sending PUBREC to subscriber_id (Mid: 88)
1414749583: Sending PUBREC to subscriber_id (Mid: 88)
1414749583: Received PUBLISH from subscriber_id (d1, q2, r0, m88, 'topic', ... (137 bytes))
1414749583: Sending PUBREC to subscriber_id (Mid: 88)
1414749583: Received SUBSCRIBE from subscriber_id
1414749583: topic/subscriber_id/# (QoS 2)
1414749583: Sending SUBACK to subscriber_id
1414749583: Received PUBREL from subscriber_id (Mid: 88)
1414749583: Sending PUBCOMP to subscriber_id (Mid: 88)
1414749583: Received PUBREL from subscriber_id (Mid: 88)
1414749583: Sending PUBCOMP to subscriber_id (Mid: 88)
1414749583: Received PUBREL from subscriber_id (Mid: 88)
1414749583: Sending PUBCOMP to subscriber_id (Mid: 88)
1414749583: Received PUBLISH from subscriber_id (d0, q2, r0, m90, 'topic', ... (137 bytes))
1414749583: Sending PUBREC to subscriber_id (Mid: 90)
1414749593: Received PINGREQ from subscriber_id
1414749593: Sending PINGRESP to subscriber_id
1414749594: Sending PUBREC to subscriber_id (Mid: 90)
----------------------------------------
Do you think I'm on the right track or do you think I'm totally wrong with this conclusion :) ?
Please tell me if you need more details.
