ClassCastException storing message in persistBufferedMessage

[albertoalcolea]

Please fill out the form below before submitting, thank you!

• Bug exists Release Version 1.2.5 ( Master Branch)
• Bug exists in MQTTv3 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)
• [?] Bug exists in MQTTv5 Client on Snapshot Version 1.2.6-SNAPSHOT (Develop Branch)

This bug relates to:

• ClassCastException when disconnected and trying to subscribe #606
• Lost connection caused by ClassCastException #622

And this fix:

• cdde804

Under some circunstances, a SUBSCRIBE or UNSUBSCRIBE message might be persisted in the persistBufferedMessage is set when the client has lost the connection with the server.

public void sendNoWait(MqttWireMessage message, MqttToken token) throws MqttException {
	final String methodName = "sendNoWait";
	if (isConnected() ||
			(!isConnected() && message instanceof MqttConnect) ||
			(isDisconnecting() && message instanceof MqttDisconnect)) {
		if(disconnectedMessageBuffer != null && disconnectedMessageBuffer.getMessageCount() != 0){
			//@TRACE 507=Client Connected, Offline Buffer available, but not empty. Adding message to buffer. message={0}
			log.fine(CLASS_NAME, methodName, "507", new Object[] {message.getKey()});
			if(disconnectedMessageBuffer.isPersistBuffer()){
				if (message instanceof MqttPublish) {
					this.clientState.persistBufferedMessage(message);
				}
			}
			disconnectedMessageBuffer.putMessage(message, token);
		} else {
			this.internalSend(message, token);
		}
	} else if(disconnectedMessageBuffer != null) {
		//@TRACE 508=Offline Buffer available. Adding message to buffer. message={0}
		log.fine(CLASS_NAME, methodName, "508", new Object[] {message.getKey()});
		if(disconnectedMessageBuffer.isPersistBuffer()){
			this.clientState.persistBufferedMessage(message);
		}
		disconnectedMessageBuffer.putMessage(message, token);
	} else {
		//@TRACE 208=failed: not connected
		log.fine(CLASS_NAME, methodName, "208");
		throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_NOT_CONNECTED);
	}
}

In this scenario, the message enters in the second branch of the conditional: else if (disconnectedMessageBuffer != null) { ...
And the code tries to persist it in the buffer, what ends up causing a ClassCastException when trying to store the message in the buffer as a MQTTPublish object in the ClientState class:

this.persistence.put(key, (MqttPublish)message);

My suggestion, which may be insufficient in case of a race condition, is to implement the same safe-check that was added in the issue 606 validating the message is an instance of the MqttPublish class:

    (...)
    } else if(disconnectedMessageBuffer != null) {
		//@TRACE 508=Offline Buffer available. Adding message to buffer. message={0}
		log.fine(CLASS_NAME, methodName, "508", new Object[] {message.getKey()});
		if(disconnectedMessageBuffer.isPersistBuffer()){
                        if (message instanceof MqttPublish) {
                            this.clientState.persistBufferedMessage(message);
                        }
		}
		disconnectedMessageBuffer.putMessage(message, token);
    } 
    (...)