Issue 549 - Trialing ignoring exceptions thrown in callbacks.

Signed-off-by: James Sutton <james.sutton@uk.ibm.com>

Author: jpwsutton

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/CommsCallback.java

@@ -264,7 +264,13 @@ private void handleActionComplete(MqttToken token) throws MqttException {
 				// If a callback is registered and delivery has finished
 				// call delivery complete callback.
 				if (mqttCallback != null && token instanceof MqttDeliveryToken && token.isComplete()) {
-					mqttCallback.deliveryComplete((MqttDeliveryToken) token);
+					try {
+						mqttCallback.deliveryComplete((MqttDeliveryToken) token);
+					} catch (Throwable ex) {
+						// Just log the fact that an exception was thrown
+						// @TRACE 726=Ignoring Exception thrown from deliveryComplete {0}
+						log.fine(CLASS_NAME, methodName, "726", new Object[] { ex });
+					}
 				}
 				// Now call async action completion callbacks
 				fireActionEvent(token);
@@ -318,9 +324,8 @@ public void connectionLost(MqttException cause, MqttDisconnect message) {
 				reconnectInternalCallback.disconnected(disconnectResponse);
 			}
 		} catch (Throwable t) {
-			// Just log the fact that a throwable has caught connection lost
-			// is called during shutdown processing so no need to do anything else
-			// @TRACE 720=exception from connectionLost {0}
+			// Just log the fact that an exception was thrown
+			// @TRACE 720=Ignoring Exception thrown from connectionLost {0}
 			log.fine(CLASS_NAME, methodName, "720", new Object[] { t });
 		}
 	}
@@ -394,8 +399,15 @@ public void messageArrived(MqttPublish sendMessage) {
 	 *            The {@link MqttAuth} message.
 	 */
 	public void authMessageReceived(MqttAuth authMessage) {
+		String methodName = "authMessageReceived";
 		if (mqttCallback != null) {
-			mqttCallback.authPacketArrived(authMessage.getReturnCode(), authMessage.getProperties());
+			try {
+				mqttCallback.authPacketArrived(authMessage.getReturnCode(), authMessage.getProperties());
+			} catch (Throwable ex) {
+				// Just log the fact that an exception was thrown
+				// @TRACE 727=Ignoring Exception thrown from authPacketArrived {0}
+				log.fine(CLASS_NAME, methodName, "727", new Object[] { ex });
+			}
 		}
 	}
 
@@ -411,7 +423,13 @@ public void mqttErrorOccurred(MqttException exception) {
 		final String methodName = "mqttErrorOccurred";
 		log.warning(CLASS_NAME, methodName, "721", new Object[] { exception.getMessage() });
 		if (mqttCallback != null) {
-			mqttCallback.mqttErrorOccurred(exception);
+			try {
+				mqttCallback.mqttErrorOccurred(exception);
+			} catch (Exception ex) {
+				// Just log the fact that an exception was thrown
+				// @TRACE 724=Ignoring Exception thrown from mqttErrorOccurred: {0}
+				log.fine(CLASS_NAME, methodName, "724", new Object[] { ex });
+			}
 		}
 	}
 
@@ -467,7 +485,7 @@ public void messageArrivedComplete(int messageId, int qos) throws MqttException
 			this.clientComms.deliveryComplete(messageId);
 			MqttPubComp pubComp = new MqttPubComp(MqttReturnCode.RETURN_CODE_SUCCESS, messageId, new MqttProperties());
 			// @TRACE 723=Creating MqttPubComp due to manual ACK: {0}
-			log.info(CLASS_NAME, "messageArrivedComplete", "723", new Object[] {pubComp.toString()});
+			log.info(CLASS_NAME, "messageArrivedComplete", "723", new Object[] { pubComp.toString() });
 
 			this.clientComms.internalSend(pubComp, new MqttToken(clientComms.getClient().getClientId()));
 		}
@@ -573,6 +591,7 @@ public void removeMessageListeners() {
 
 	protected boolean deliverMessage(String topicName, int messageId, MqttMessage aMessage) throws Exception {
 		boolean delivered = false;
+		String methodName = "deliverMessage";
 
 		if (aMessage.getProperties().getSubscriptionIdentifiers().isEmpty()) {
 			// No Subscription IDs, use topic filter matching
@@ -602,7 +621,13 @@ protected boolean deliverMessage(String topicName, int messageId, MqttMessage aM
 		 */
 		if (mqttCallback != null && !delivered) {
 			aMessage.setId(messageId);
-			mqttCallback.messageArrived(topicName, aMessage);
+			try {
+				mqttCallback.messageArrived(topicName, aMessage);
+			} catch (Exception ex) {
+				// Just log the fact that an exception was thrown
+				// @TRACE 725=Ignoring Exception thrown from messageArrived: {0}
+				log.fine(CLASS_NAME, methodName, "725", new Object[] { ex });
+			}
 			delivered = true;
 		}
 

org.eclipse.paho.mqttv5.client/src/main/java/org/eclipse/paho/mqttv5/client/internal/ConnectActionListener.java

@@ -159,7 +159,12 @@ public void onSuccess(IMqttToken token) {
 
 		if (mqttCallback != null) {
 			String serverURI = comms.getNetworkModules()[comms.getNetworkModuleIndex()].getServerURI();
-			mqttCallback.connectComplete(reconnect, serverURI);
+			try {
+				mqttCallback.connectComplete(reconnect, serverURI);
+			} catch (Throwable ex) {
+				// Just catch any exceptions thrown here and ignore.
+			}
+
 		}
 
 	}

org.eclipse.paho.mqttv5.client/src/main/resources/org/eclipse/paho/mqttv5/client/internal/nls/logcat.properties

@@ -162,10 +162,16 @@
 716=call onSuccess key={0}
 717=call onFailure key {0}
 719=callback threw ex:
-720=exception from connectionLost {0}
+720=Ignoring Exception thrown from connectionLost {0}
+725=Ignoring Exception thrown from messageArrived: {0}
+726=726=Ignoring Exception thrown from deliveryComplete {0}
 721=Non-Critical MQTT error thrown, passing back to application={0}
 722=Server initiated disconnect, connection closed. Disconnect={0}
 723=Creating MqttPubComp due to manual ACK: {0}
+724=Ignoring Exception thrown from mqttErrorOccurred: {0}
+725=Ignoring Exception thrown from messageArrived: {0}
+726=Ignoring Exception thrown from deliveryComplete {0}
+727=Ignoring Exception thrown from authPacketArrived {0}
 800=stopping sender
 801=stopped
 802=network send key={0} msg={1}

org.eclipse.paho.mqttv5.client/src/test/java/org/eclipse/paho/mqttv5/client/test/BasicTest.java

@@ -8,8 +8,6 @@
 import org.eclipse.paho.mqttv5.client.IMqttDeliveryToken;
 import org.eclipse.paho.mqttv5.client.IMqttToken;
 import org.eclipse.paho.mqttv5.client.MqttAsyncClient;
-import org.eclipse.paho.mqttv5.client.MqttCallback;
-import org.eclipse.paho.mqttv5.client.MqttDisconnectResponse;
 import org.eclipse.paho.mqttv5.client.test.logging.LoggingUtilities;
 import org.eclipse.paho.mqttv5.client.test.properties.TestProperties;
 import org.eclipse.paho.mqttv5.client.test.utilities.MqttV5Receiver;
@@ -18,7 +16,6 @@
 import org.eclipse.paho.mqttv5.common.MqttException;
 import org.eclipse.paho.mqttv5.common.MqttMessage;
 import org.eclipse.paho.mqttv5.common.MqttSubscription;
-import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
 import org.junit.Assert;
 import org.junit.BeforeClass;
 import org.junit.Test;