Added removing message function. (#396)

Signed-off-by: Takatoshi Kondo <redboltz@gmail.com>

Author: redboltz

org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/SendReceiveAsyncCallbackTest.java

@@ -153,15 +153,18 @@ public MqttMessage getNextMessage() {
 
 		public void messageArrived(String topic, MqttMessage message)
 				throws Exception {
-
-			log.info("message arrived: '" + new String(message.getPayload())
+			String msgstr = new String(message.getPayload());
+			log.info("message arrived: '" + msgstr
 					+ "' " + this.hashCode() + " "
 					+ (message.isDuplicate() ? "duplicate" : ""));
 
 			if (!message.isDuplicate()) {
 				synchronized (messages) {
-					messages.add(message);
-					messages.notifyAll();
+					if (!msgstr.equals("might cancel")) {
+						log.info("add message");
+						messages.add(message);
+						messages.notifyAll();
+					}
 				}
 			}
 		}
@@ -190,6 +193,16 @@ public void onSuccess(IMqttToken token) {
 						token.getClient().publish(topicFilter, "my data".getBytes(), 2, false, null, myOnPublish);
 					}
 					else {
+						IMqttDeliveryToken tokenToRemove1 = token.getClient().publish(topicFilter, "might cancel".getBytes(), 1, false, null, myOnPublish);
+						boolean res1_1 = token.getClient().removeMessage(tokenToRemove1);
+						Assert.assertTrue("message (QoS1) removed", res1_1);
+						boolean res1_2 = token.getClient().removeMessage(tokenToRemove1);
+						Assert.assertFalse("already removed message (QoS1) shoudn't be removed", res1_2);
+						IMqttDeliveryToken tokenToRemove2 = token.getClient().publish(topicFilter, "might cancel".getBytes(), 2, false, null, myOnPublish);
+						boolean res2_1 = token.getClient().removeMessage(tokenToRemove2);
+						Assert.assertTrue("message (QoS2) removed", res2_1);
+						boolean res2_2 = token.getClient().removeMessage(tokenToRemove2);
+						Assert.assertFalse("already removed message (QoS2) shoudn't be removed", res2_2);
 						log.info(methodName + ": all messages published");
 						testFinished = true;
 					}

org.eclipse.paho.client.mqttv3/src/main/java-templates/org/eclipse/paho/client/mqttv3/internal/ClientComms.java

@@ -29,6 +29,7 @@
 import org.eclipse.paho.client.mqttv3.IMqttActionListener;
 import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
 import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttCallback;
 import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
 import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
@@ -200,6 +201,17 @@ public void sendNoWait(MqttWireMessage message, MqttToken token) throws MqttExce
 		}
 	}
 
+	/**
+	 * Removes the message corresponding to the token from the outbound queue and persistence.
+	 * @param token The {@link IMqttDeliveryMqttToken} to remove
+	 * @return if the message is removed, then true, otherwise false
+	 * @throws MqttException if an error occurs sending the message
+	 */
+	public boolean removeMessage(IMqttDeliveryToken token) throws MqttException {
+		final String methodName = "removeMessage";
+		return this.clientState.removeMessage(token);
+	}
+
 	/**
 	 * Close and tidy up.
 	 *

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/IMqttAsyncClient.java

@@ -809,6 +809,22 @@ public IMqttToken unsubscribe(String topicFilter, Object userContext, IMqttActio
 	public IMqttToken unsubscribe(String[] topicFilters, Object userContext, IMqttActionListener callback)
 			throws MqttException;
 
+	/**
+	 * Removes a published message corresponding to the token.
+	 * <p>If a publish is requested with QoS1 or Qos2 and the publish callback is
+	 * not called yet, this function returns true, the publish called will never
+	 * be called, and a messageId corresponding to the token will become reusable.
+	 * </p>
+	 * <p>If the publish callback is already be called, this function returns false.
+	 * </p>
+	 * <p>This function might not stop sending the published message.
+	 * </p>
+	 * 	 *
+	 * @param token the token of removing published message
+	 * @return if the message is removed then true, otherwise false
+	 * @throws MqttException if there was an error removing the message.
+	 */
+	public boolean removeMessage(IMqttDeliveryToken token) throws MqttException;
 
 	/**
 	 * Sets a callback listener to use for events that happen asynchronously.

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java

@@ -1237,6 +1237,15 @@ public IMqttToken unsubscribe(String[] topicFilters, Object userContext, IMqttAc
 		return token;
 	}
 
+	/*
+	 * (non-Javadoc)
+	 * 
+	 * @see IMqttAsyncClient#removeMessage(IMqttDeliveryToken)
+	 */
+	public boolean removeMessage(IMqttDeliveryToken token) throws MqttException {
+		return comms.removeMessage(token);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * 

org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/ClientState.java

@@ -28,6 +28,7 @@
 import java.util.Vector;
 
 import org.eclipse.paho.client.mqttv3.IMqttActionListener;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
 import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
 import org.eclipse.paho.client.mqttv3.MqttException;
@@ -191,6 +192,10 @@ private String getSendPersistenceKey(MqttWireMessage message) {
 		return PERSISTENCE_SENT_PREFIX + message.getMessageId();
 	}
 
+	private String getSendPersistenceKey(int messageId) {
+		return PERSISTENCE_SENT_PREFIX + messageId;
+	}
+	
 	private String getSendConfirmPersistenceKey(MqttWireMessage message) {
 		return PERSISTENCE_CONFIRMED_PREFIX + message.getMessageId();
 	}
@@ -641,6 +646,39 @@ protected void undo(MqttPublish message) throws MqttPersistenceException {
 		}
 	}
 
+	/**
+	 * This removes the MqttSend message from the outbound queue and persistence.
+	 * @param message The {@link MqttPublish} message to remove
+	 * @return if the message is removed, then true, otherwise false
+	 * @throws MqttException if an exception occurs whilst removing the message
+	 */
+	protected boolean removeMessage(IMqttDeliveryToken token) throws MqttException {
+		final String methodName = "removeMessage";
+		MqttMessage message = token.getMessage();
+		int messageId = token.getMessageId();
+		boolean result = false;
+		synchronized (queueLock) {
+			if (message.getQos() == 1) {
+				if (outboundQoS1.remove(new Integer(messageId)) != null) {
+					result = true;
+				}
+			}
+			if (message.getQos() == 2) {
+				if (outboundQoS2.remove(new Integer(messageId)) != null) {
+					result = true;
+				}
+			}
+			if (pendingMessages.removeElement(message)) {
+				result = true;
+			}
+			persistence.remove(getSendPersistenceKey(messageId));
+			String key = new Integer(messageId).toString();
+			tokenStore.removeToken(key);
+			releaseMessageId(messageId);
+		}
+		return result;
+	}
+	
 	/**
 	 * Check and send a ping if needed and check for ping timeout.
 	 * Need to send a ping if nothing has been sent or received  
@@ -861,6 +899,7 @@ protected void notifySent(MqttWireMessage message) {
 		log.fine(CLASS_NAME,methodName,"625",new Object[]{message.getKey()});
 
 		MqttToken token = tokenStore.getToken(message);
+		if (token == null) return;
 		token.internalTok.notifySent();
         if (message instanceof MqttPingReq) {
             synchronized (pingOutstandingLock) {