Merge remote-tracking branch 'upstream/develop' into patch-3

Author: Ing. Jan Kaláb

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

@@ -60,10 +60,18 @@ public void testValidTopicFilterWildcards() throws Exception {
 	public void testMatchedTopicFilterWildcards() throws Exception {
 		String methodName = Utility.getMethodName();
 		LoggingUtilities.banner(log, cclass, methodName);
-		String[][] matchingTopics = new String[][] { { "sport/tennis/player1/#", "sport/tennis/player1" },
-				{ "sport/tennis/player1/#", "sport/tennis/player1/ranking" },
-				{ "sport/tennis/player1/#", "sport/tennis/player1/score/wimbledon" }, { "sport/#", "sport" },
-				{ "#", "sport/tennis/player1" } };
+		String[][] matchingTopics = new String[][] {
+                        { "+/+", "sport/hockey" },
+                        { "/+", "/sport" },
+                        { "sport/tennis/player1/#", "sport/tennis/player1" },
+                        { "sport/tennis/player1/#", "sport/tennis/player1/ranking" },
+                        { "sport/tennis/player1/#", "sport/tennis/player1/score/wimbledon" },
+                        { "sport/#", "sport" },
+                        { "#", "sport/tennis/player1" },
+                        { "sport/tennis/player1/#", "sport/tennis/player1//wimbledon" },
+                        { "sport/+/player1/#", "sport/tennis/player1/wimbledon" },
+                        { "sport/+/player1/#", "sport/soccer/player1/UEFA" }
+                        };
 
 		for (String[] pair : matchingTopics) {
 			Assert.assertTrue(pair[0] + " should match " + pair[1], MqttTopic.isMatched(pair[0], pair[1]));
@@ -74,8 +82,15 @@ public void testMatchedTopicFilterWildcards() throws Exception {
 	public void testNonMatchedTopicFilterWildcards() throws Exception {
 		String methodName = Utility.getMethodName();
 		LoggingUtilities.banner(log, cclass, methodName);
-		String[][] matchingTopics = new String[][] { { "sport/tennis/player1/#", "sport/tennis/player2" },
-				{ "sport1/#", "sport2" }, { "sport/tennis1/player/#", "sport/tennis2/player" } };
+		String[][] matchingTopics = new String[][] {
+                        { "+/+", "/sport" },
+                        { "+/+", "a/b/c" },
+                        { "/sport/+", "/sport/" },
+                        { "sport/tennis/player1/#", "sport/tennis/player2" },
+                        { "sport1/#", "sport2" },
+                        { "sport/tennis1/player/#", "sport/tennis2/player" },
+                        { "sport//tennis/player1/#", "sport/tennis/player1//wimbledon" }
+                        };
 
 		for (String[] pair : matchingTopics) {
 			Assert.assertFalse(pair[0] + " should NOT match " + pair[1], MqttTopic.isMatched(pair[0], pair[1]));

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

@@ -82,7 +82,7 @@ public static void setUpBeforeClass() throws Exception {
 	 * 
 	 * @throws Exception
 	 */
-	@Test(timeout=30000)
+	@Test(timeout=60000)
 	public void testSSLSessionInvalidated() throws Exception {
 		 //System.setProperty("javax.net.debug", "all");
 

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

@@ -275,6 +275,9 @@ public static boolean isMatched(String topicFilter, String topicName) throws Ill
 		}
 
 		while (filterPos < filterLen && topicPos < topicLen) {
+                        if (topicFilter.charAt(filterPos) == '#')
+                                topicPos = topicLen - 1; // skip until end of string
+
 			if (topicName.charAt(topicPos) == '/' && topicFilter.charAt(filterPos) != '/')
 
 				break;
@@ -285,8 +288,8 @@ public static boolean isMatched(String topicFilter, String topicName) throws Ill
 				int nextpos = topicPos + 1;
 				while (nextpos < topicLen && topicName.charAt(nextpos) != '/')
 					nextpos = ++topicPos + 1;
-			} else if (topicFilter.charAt(filterPos) == '#')
-				topicPos = topicLen - 1; // skip until end of string
+			}
+
 			filterPos++;
 			topicPos++;
 		}

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

@@ -119,7 +119,7 @@ public class ClientState {
 	private CommsTokenStore tokenStore;
 	private ClientComms clientComms = null;
 	private CommsCallback callback = null;
-	private long keepAlive;
+	private long keepAliveNanos;					// nanoseconds time
 	private boolean cleanSession;
 	private MqttClientPersistence persistence;
 
@@ -131,9 +131,9 @@ public class ClientState {
 	private final Object quiesceLock = new Object();
 	private boolean quiescing = false;
 
-	private long lastOutboundActivity = 0;
-	private long lastInboundActivity = 0;
-	private long lastPing = 0;
+	private long lastOutboundActivity = 0;			// nanoseconds absolute time
+	private long lastInboundActivity = 0;			// nanoseconds absolute time
+	private long lastPing = 0;						// nanoseconds absolute time
 	private MqttWireMessage pingCommand;
 	private final Object pingOutstandingLock = new Object();
 	private int pingOutstanding = 0;
@@ -177,14 +177,14 @@ protected void setMaxInflight(int maxInflight) {
         pendingMessages = new Vector(this.maxInflight);
     }
     protected void setKeepAliveSecs(long keepAliveSecs) {
-		this.keepAlive = TimeUnit.SECONDS.toNanos(keepAliveSecs);
+		this.keepAliveNanos = TimeUnit.SECONDS.toNanos(keepAliveSecs);
 	}
     /**
      * Returns the keepAlive in Milliseconds
      * @return The KeepAlive value in Milliseconds
      */
 	protected long getKeepAlive() {
-		return TimeUnit.NANOSECONDS.toMillis(this.keepAlive);
+		return TimeUnit.NANOSECONDS.toMillis(this.keepAliveNanos);
 	}
 	protected void setCleanSession(boolean cleanSession) {
 		this.cleanSession = cleanSession;
@@ -715,9 +715,9 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
         }
 
 		MqttToken token = null;
-		long nextPingTime = this.keepAlive;
+		long nextPingTime = TimeUnit.NANOSECONDS.toMillis(this.keepAliveNanos);		// milliseconds relative time
 
-		if (connected && this.keepAlive > 0) {
+		if (connected && this.keepAliveNanos > 0) {
 			long time = System.nanoTime();
 			// Below might not be necessary since move to nanoTime (Issue #278)
 			//Reduce schedule frequency since System.currentTimeMillis is no accurate, add a buffer
@@ -728,24 +728,24 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
             synchronized (pingOutstandingLock) {
 
                 // Is the broker connection lost because the broker did not reply to my ping?                                                                                                                                 
-                if (pingOutstanding > 0 && (time - lastInboundActivity >= keepAlive + delta)) {
+                if (pingOutstanding > 0 && (time - lastInboundActivity >= keepAliveNanos + delta)) {
                     // lastInboundActivity will be updated once receiving is done.                                                                                                                                        
                     // Add a delta, since the timer and System.currentTimeMillis() is not accurate.     
                 		// TODO - Remove Delta, maybe?
                 	// A ping is outstanding but no packet has been received in KA so connection is deemed broken                                                                                                         
                     //@TRACE 619=Timed out as no activity, keepAlive={0} lastOutboundActivity={1} lastInboundActivity={2} time={3} lastPing={4}                                                                           
-                    log.severe(CLASS_NAME,methodName,"619", new Object[]{ Long.valueOf(this.keepAlive), Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity),  Long.valueOf(time),  Long.valueOf(lastPing)});
+                    log.severe(CLASS_NAME,methodName,"619", new Object[]{ Long.valueOf(this.keepAliveNanos), Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity),  Long.valueOf(time),  Long.valueOf(lastPing)});
 
                     // A ping has already been sent. At this point, assume that the                                                                                                                                       
                     // broker has hung and the TCP layer hasn't noticed.                                                                                                                                                  
                     throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_TIMEOUT);
                 }
 
                 // Is the broker connection lost because I could not get any successful write for 2 keepAlive intervals?                                                                                                      
-                if (pingOutstanding == 0 && (time - lastOutboundActivity >= 2*keepAlive)) {
+                if (pingOutstanding == 0 && (time - lastOutboundActivity >= 2* keepAliveNanos)) {
 
                     // I am probably blocked on a write operations as I should have been able to write at least a ping message                                                                                                    
-                	log.severe(CLASS_NAME,methodName,"642", new Object[]{ Long.valueOf(this.keepAlive), Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity),  Long.valueOf(time),  Long.valueOf(lastPing)});
+                	log.severe(CLASS_NAME,methodName,"642", new Object[]{ Long.valueOf(this.keepAliveNanos), Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity),  Long.valueOf(time),  Long.valueOf(lastPing)});
 
                     // A ping has not been sent but I am not progressing on the current write operation. 
                 	// At this point, assume that the broker has hung and the TCP layer hasn't noticed.                                                                                                                                                  
@@ -761,11 +761,11 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
                 //    This would be the case when receiving a large message;                                                                                                                                                  
                 //    the broker needs to keep receiving a regular ping even if the ping response are queued after the long message                                                                                           
                 //    If lacking to do so, the broker will consider my connection lost and cut my socket.                                                                                                                     
-                if ((pingOutstanding == 0 && (time - lastInboundActivity >= keepAlive - delta)) ||
-                    (time - lastOutboundActivity >= keepAlive - delta)) {
+                if ((pingOutstanding == 0 && (time - lastInboundActivity >= keepAliveNanos - delta)) ||
+                    (time - lastOutboundActivity >= keepAliveNanos - delta)) {
 
                     //@TRACE 620=ping needed. keepAlive={0} lastOutboundActivity={1} lastInboundActivity={2}                                                                                                              
-                    log.fine(CLASS_NAME,methodName,"620", new Object[]{ Long.valueOf(this.keepAlive), Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity)});
+                    log.fine(CLASS_NAME,methodName,"620", new Object[]{ Long.valueOf(this.keepAliveNanos), Long.valueOf(lastOutboundActivity), Long.valueOf(lastInboundActivity)});
 
                     // pingOutstanding++;  // it will be set after the ping has been written on the wire                                                                                                             
                     // lastPing = time;    // it will be set after the ping has been written on the wire                                                                                                             
@@ -784,7 +784,9 @@ public MqttToken checkForActivity(IMqttActionListener pingCallback) throws MqttE
                 else {
                 		//@TRACE 634=ping not needed yet. Schedule next ping.
                     log.fine(CLASS_NAME, methodName, "634", null);
-                    nextPingTime = Math.max(1,  getKeepAlive() - (time - lastOutboundActivity));
+                    long elapsedSinceLastActivityNanos = time - lastOutboundActivity;
+                    long elapsedSinceLastActivityMillis = TimeUnit.NANOSECONDS.toMillis( elapsedSinceLastActivityNanos );
+                    nextPingTime = Math.max(1,  getKeepAlive() - elapsedSinceLastActivityMillis);
                 }
             }
             //@TRACE 624=Schedule next ping at {0}                                                                                                                                                                                
@@ -882,8 +884,13 @@ protected MqttWireMessage get() throws MqttException {
 		return result;
 	}
 
+	/**
+	 * Sets the keep alive interval.
+	 *
+	 * @param interval keep alive interval in milliseconds.
+	 */
 	public void setKeepAliveInterval(long interval) {
-		this.keepAlive = interval;
+		this.keepAliveNanos = TimeUnit.MILLISECONDS.toNanos(interval);
 	}
 
     public void notifySentBytes(int sentBytesCount) {

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

@@ -165,7 +165,13 @@ public void run() {
 						if (message != null) {
 							// A new message has arrived
 							clientState.notifyReceivedMsg(message);
-						}
+						}  
+                                                else {
+                                                    // fix for bug 719
+                                                    if (!clientComms.isConnected()) {
+                                                         throw new IOException("Connection is lost.");
+                                                    }
+                                                }
 					}
 				}
 				catch (MqttException ex) {

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

@@ -27,7 +27,7 @@
 
 public class DisconnectedMessageBuffer implements Runnable {
 
-	private final String CLASS_NAME = "DisconnectedMessageBuffer";
+	private final String CLASS_NAME = DisconnectedMessageBuffer.class.getName();
 	private Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 	private DisconnectedBufferOptions bufferOpts;
 	private ArrayList<BufferedMessage> buffer;
@@ -53,6 +53,11 @@ public DisconnectedMessageBuffer(DisconnectedBufferOptions options) {
 	 *             if the Buffer is full
 	 */
 	public void putMessage(MqttWireMessage message, MqttToken token) throws MqttException {
+		if (token != null) {
+			message.setToken(token);
+			token.internalTok.setMessageID(message.getMessageId());
+		}
+		
 		BufferedMessage bufferedMessage = new BufferedMessage(message, token);
 		synchronized (bufLock) {
 			if (buffer.size() < bufferOpts.getBufferSize()) {

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

@@ -28,7 +28,7 @@
 
 public class DisconnectedMessageBuffer implements Runnable {
 
-	private static final String CLASS_NAME = "DisconnectedMessageBuffer";
+	private static final String CLASS_NAME = DisconnectedMessageBuffer.class.getName();
 	private Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME);
 	private DisconnectedBufferOptions bufferOpts;
 	private ArrayList<BufferedMessage> buffer;