Fixes #861: Perfomance degradation with increasing subscription count.

Iterating over all subscriptions for each incoming message is really
slow when there are many subscriptions.
There can only ever be one Subscription with no wildcards that matches a
message topic, so iterating over all such subscriptions is not needed. A
simple map lookup is enough. Wildcard subscriptions do need to be
iterated over, since there is no easy way to see which match and witch
do not.

Signed-off-by: Hylke van der Schaaf <hylke.vds@gmail.com>

Author: hylkevds

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

@@ -52,8 +52,9 @@ public class CommsCallback implements Runnable {
 	private static final int INBOUND_QUEUE_SIZE = 10;
 	private MqttCallback mqttCallback;
 	private MqttCallbackExtended reconnectInternalCallback;
-	private Hashtable<String, IMqttMessageListener> callbacks; // topicFilter -> messageHandler
-	private ClientComms clientComms;
+	private final Hashtable<String, IMqttMessageListener> callbacksWildcards; // topicFilter with wildcards -> messageHandler
+	private final Hashtable<String, IMqttMessageListener> callbacksDirect; // topicFilter without wildcards -> messageHandler
+	private final ClientComms clientComms;
 	private final Vector<MqttWireMessage> messageQueue;
 	private final Vector<MqttToken> completeQueue;
 
@@ -75,7 +76,8 @@ private enum State {STOPPED, RUNNING, QUIESCING}
 		this.clientComms = clientComms;
 		this.messageQueue = new Vector<MqttWireMessage>(INBOUND_QUEUE_SIZE);
 		this.completeQueue = new Vector<MqttToken>(INBOUND_QUEUE_SIZE);
-		this.callbacks = new Hashtable<String, IMqttMessageListener>();
+		this.callbacksWildcards = new Hashtable<String, IMqttMessageListener>();
+		this.callbacksDirect = new Hashtable<String, IMqttMessageListener>();
 		log.setResourceName(clientComms.getClient().getClientId());
 	}
 
@@ -351,7 +353,7 @@ public void fireActionEvent(MqttToken token) {
 	 */
 	public void messageArrived(MqttPublish sendMessage) {
 		final String methodName = "messageArrived";
-		if (mqttCallback != null || callbacks.size() > 0) {
+		if (mqttCallback != null || !callbacksWildcards.isEmpty() || !callbacksDirect.isEmpty()) {
 			// If we already have enough messages queued up in memory, wait
 			// until some more queue space becomes available. This helps 
 			// the client protect itself from getting flooded by messages 
@@ -481,34 +483,47 @@ protected Thread getThread() {
 
 
 	public void setMessageListener(String topicFilter, IMqttMessageListener messageListener) {
-		this.callbacks.put(topicFilter, messageListener);
+		if (topicFilter.contains("#") || topicFilter.contains("+")) {
+			this.callbacksWildcards.put(topicFilter, messageListener);
+		} else {
+			this.callbacksDirect.put(topicFilter, messageListener);
+		}
 	}
 
 
 	public void removeMessageListener(String topicFilter) {
-		this.callbacks.remove(topicFilter); // no exception thrown if the filter was not present
+		this.callbacksWildcards.remove(topicFilter); // no exception thrown if the filter was not present
+		this.callbacksDirect.remove(topicFilter); // no exception thrown if the filter was not present
 	}
 
 	public void removeMessageListeners() {
-		this.callbacks.clear(); 
+		this.callbacksWildcards.clear();
+		this.callbacksWildcards.clear();
 	}
 
 
 	protected boolean deliverMessage(String topicName, int messageId, MqttMessage aMessage) throws Exception
 	{		
 		boolean delivered = false;
 
-		Enumeration<String> keys = callbacks.keys();
+		IMqttMessageListener callback = this.callbacksDirect.get(topicName);
+		if (callback != null) {
+			aMessage.setId(messageId);
+			callback.messageArrived(topicName, aMessage);
+			delivered = true;
+		}
+		
+		Enumeration<String> keys = callbacksWildcards.keys();
 		while (keys.hasMoreElements()) {
 			String topicFilter = (String)keys.nextElement();
 			// callback may already have been removed in the meantime, so a null check is necessary
-			IMqttMessageListener callback = callbacks.get(topicFilter);
+			callback = callbacksWildcards.get(topicFilter);
 			if(callback == null) {
 				continue;
 			}
 			if (MqttTopic.isMatched(topicFilter, topicName)) {
 				aMessage.setId(messageId);
-				((IMqttMessageListener)callback).messageArrived(topicName, aMessage);
+				callback.messageArrived(topicName, aMessage);
 				delivered = true;
 			}
 		}