From 54f110d05cfb626fdd704569ba0c4ed90fb8e92a Mon Sep 17 00:00:00 2001 From: James Sutton Date: Tue, 25 Jul 2017 09:35:40 +0100 Subject: [PATCH] Test and Client fixes (#385) * Resolving Issue #384 - Incorrect Log message * Issue #384 - Mapping correct Logcat Message * Updating Test Properties to point at local broker * Issue #383 - Delete MqttDefaultFilePersistence directory on clear() * Issue #356 - Fixed issue where hostnames containing underscores would not be parsed correctly by Java * Adding Coveralls integration * Issue #353 - Adding missing EDL reference to copyright statements * Updating Java Version for Mqtt V3 client eclipse project Signed-off-by: James Sutton --- .../paho/client/eclipse/view/Activator.java | 10 +- .../client/eclipse/view/ClientConstants.java | 10 +- .../client/eclipse/view/MqttClientView.java | 10 +- .../paho/client/mqttv3/test/Issue370Test.java | 9 +- .../mqttv3/test/SSLSessionResumptionTest.java | 12 +- .../src/test/resources/test.properties | 6 +- .../.settings/org.eclipse.jdt.core.prefs | 11 +- .../paho/client/mqttv3/IMqttToken.java | 10 +- .../paho/client/mqttv3/MqttAsyncClient.java | 1269 ++++++++++------- .../internal/wire/MqttOutputStream.java | 4 +- .../persist/MqttDefaultFilePersistence.java | 1 + pom.xml | 5 + 12 files changed, 820 insertions(+), 537 deletions(-) diff --git a/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/Activator.java b/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/Activator.java index 5c8a08c49..bf3d7ad74 100644 --- a/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/Activator.java +++ b/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/Activator.java @@ -1,10 +1,14 @@ -/* +/******************************************************************************* * Copyright (c) 2012 Eurotech Inc. All rights reserved. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 - * which accompanies this distribution, and is available at - * http://www.eclipse.org/legal/epl-v10.html + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Chad Kienle diff --git a/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/ClientConstants.java b/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/ClientConstants.java index 0da7f4fdd..c684c578b 100644 --- a/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/ClientConstants.java +++ b/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/ClientConstants.java @@ -1,10 +1,14 @@ -/* +/******************************************************************************* * Copyright (c) 2012 Eurotech Inc. All rights reserved. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 - * which accompanies this distribution, and is available at - * http://www.eclipse.org/legal/epl-v10.html + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Chad Kienle diff --git a/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/MqttClientView.java b/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/MqttClientView.java index c96e27be4..990a4571a 100644 --- a/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/MqttClientView.java +++ b/org.eclipse.paho.client.eclipse.view/src/org/eclipse/paho/client/eclipse/view/MqttClientView.java @@ -1,10 +1,14 @@ -/* +/******************************************************************************* * Copyright (c) 2012 Eurotech Inc. All rights reserved. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 - * which accompanies this distribution, and is available at - * http://www.eclipse.org/legal/epl-v10.html + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Chad Kienle diff --git a/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/Issue370Test.java b/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/Issue370Test.java index f4f2fb2d8..3e529b439 100644 --- a/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/Issue370Test.java +++ b/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/Issue370Test.java @@ -1,9 +1,14 @@ /******************************************************************************* * Copyright (c) 2017 Red Hat Inc. + * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 - * which accompanies this distribution, and is available at - * http://www.eclipse.org/legal/epl-v10.html + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Red Hat Inc - initial API and implementation diff --git a/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/SSLSessionResumptionTest.java b/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/SSLSessionResumptionTest.java index 347b69299..b3f9d196b 100644 --- a/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/SSLSessionResumptionTest.java +++ b/org.eclipse.paho.client.mqttv3.test/src/test/java/org/eclipse/paho/client/mqttv3/test/SSLSessionResumptionTest.java @@ -1,10 +1,14 @@ /** * Copyright (c) 2011, 2014 Eurotech and/or its affiliates * - * All rights reserved. This program and the accompanying materials - * are made available under the terms of the Eclipse Public License v1.0 - * which accompanies this distribution, and is available at - * http://www.eclipse.org/legal/epl-v10.html + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Cristiano De Alti - Eurotech (Initial contribution) diff --git a/org.eclipse.paho.client.mqttv3.test/src/test/resources/test.properties b/org.eclipse.paho.client.mqttv3.test/src/test/resources/test.properties index 833c07635..7688045b5 100644 --- a/org.eclipse.paho.client.mqttv3.test/src/test/resources/test.properties +++ b/org.eclipse.paho.client.mqttv3.test/src/test/resources/test.properties @@ -1,12 +1,12 @@ # This is the server URI which will be set in the constructor of an MQTT Client # The default is "tcp://:1883" with expressed in IPV4 dotted decimal notation -SERVER_URI=tcp://iot.eclipse.org:1883 +SERVER_URI=tcp://localhost:1883 CLIENT_KEY_STORE=clientkeystore.jks CLIENT_KEY_STORE_PASSWORD=password CLIENT_TRUST_STORE=clientkeystore.jks SERVER_SSL_PORT=18884 -SERVER_WEBSOCKET_URI=wss://iot.eclipse.org:443 -SERVER_SSL_PORT=8883 +SERVER_WEBSOCKET_URI=ws://localhost:8080 + # The list of server URIs which may be set in the MQTT ConnectOptions for an HA testcase. # There is no default value diff --git a/org.eclipse.paho.client.mqttv3/.settings/org.eclipse.jdt.core.prefs b/org.eclipse.paho.client.mqttv3/.settings/org.eclipse.jdt.core.prefs index 008a2612d..9c4403fe9 100644 --- a/org.eclipse.paho.client.mqttv3/.settings/org.eclipse.jdt.core.prefs +++ b/org.eclipse.paho.client.mqttv3/.settings/org.eclipse.jdt.core.prefs @@ -1,12 +1,13 @@ eclipse.preferences.version=1 org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled -org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.2 +org.eclipse.jdt.core.compiler.codegen.methodParameters=do not generate +org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.7 org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve -org.eclipse.jdt.core.compiler.compliance=1.4 +org.eclipse.jdt.core.compiler.compliance=1.7 org.eclipse.jdt.core.compiler.debug.lineNumber=generate org.eclipse.jdt.core.compiler.debug.localVariable=generate org.eclipse.jdt.core.compiler.debug.sourceFile=generate -org.eclipse.jdt.core.compiler.problem.assertIdentifier=warning -org.eclipse.jdt.core.compiler.problem.enumIdentifier=warning +org.eclipse.jdt.core.compiler.problem.assertIdentifier=error +org.eclipse.jdt.core.compiler.problem.enumIdentifier=error org.eclipse.jdt.core.compiler.problem.forbiddenReference=warning -org.eclipse.jdt.core.compiler.source=1.3 +org.eclipse.jdt.core.compiler.source=1.7 diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/IMqttToken.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/IMqttToken.java index 4589f95fa..d19ccde17 100644 --- a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/IMqttToken.java +++ b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/IMqttToken.java @@ -1,10 +1,14 @@ -/* +/************************************************************************** * Copyright (c) 2009, 2012 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 - * which accompanies this distribution, and is available at - * http://www.eclipse.org/legal/epl-v10.html + * and Eclipse Distribution License v1.0 which accompany this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * and the Eclipse Distribution License is available at + * http://www.eclipse.org/org/documents/edl-v10.php. * * Contributors: * Dave Locke - initial API and implementation and/or initial documentation diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java index 19e7fd37a..ae5150b6b 100644 --- a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java +++ b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java @@ -21,6 +21,7 @@ package org.eclipse.paho.client.mqttv3; +import java.lang.reflect.Field; import java.net.URI; import java.net.URISyntaxException; import java.util.Hashtable; @@ -57,39 +58,49 @@ * Lightweight client for talking to an MQTT server using non-blocking methods * that allow an operation to run in the background. * - *

This class implements the non-blocking {@link IMqttAsyncClient} client interface - * allowing applications to initiate MQTT actions and then carry on working while the - * MQTT action completes on a background thread. - * This implementation is compatible with all Java SE runtimes from 1.7 and up. + *

+ * This class implements the non-blocking {@link IMqttAsyncClient} client + * interface allowing applications to initiate MQTT actions and then carry on + * working while the MQTT action completes on a background thread. This + * implementation is compatible with all Java SE runtimes from 1.7 and up. + *

+ *

+ * An application can connect to an MQTT server using: *

- *

An application can connect to an MQTT server using:

*
    *
  • A plain TCP socket *
  • A secure SSL/TLS socket *
* - *

To enable messages to be delivered even across network and client restarts - * messages need to be safely stored until the message has been delivered at the requested - * quality of service. A pluggable persistence mechanism is provided to store the messages. + *

+ * To enable messages to be delivered even across network and client restarts + * messages need to be safely stored until the message has been delivered at the + * requested quality of service. A pluggable persistence mechanism is provided + * to store the messages. *

- *

By default {@link MqttDefaultFilePersistence} is used to store messages to a file. - * If persistence is set to null then messages are stored in memory and hence can be lost - * if the client, Java runtime or device shuts down. + *

+ * By default {@link MqttDefaultFilePersistence} is used to store messages to a + * file. If persistence is set to null then messages are stored in memory and + * hence can be lost if the client, Java runtime or device shuts down. *

- *

If connecting with {@link MqttConnectOptions#setCleanSession(boolean)} set to true it - * is safe to use memory persistence as all state is cleared when a client disconnects. If - * connecting with cleanSession set to false in order to provide reliable message delivery - * then a persistent message store such as the default one should be used. + *

+ * If connecting with {@link MqttConnectOptions#setCleanSession(boolean)} set to + * true it is safe to use memory persistence as all state is cleared when a + * client disconnects. If connecting with cleanSession set to false in order to + * provide reliable message delivery then a persistent message store such as the + * default one should be used. *

- *

The message store interface is pluggable. Different stores can be used by implementing - * the {@link MqttClientPersistence} interface and passing it to the clients constructor. + *

+ * The message store interface is pluggable. Different stores can be used by + * implementing the {@link MqttClientPersistence} interface and passing it to + * the clients constructor. *

* * @see IMqttAsyncClient */ public class MqttAsyncClient implements IMqttAsyncClient { private static final String CLASS_NAME = MqttAsyncClient.class.getName(); - private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT,CLASS_NAME); + private static final Logger log = LoggerFactory.getLogger(LoggerFactory.MQTT_CLIENT_MSG_CAT, CLASS_NAME); private static final String CLIENT_ID_PREFIX = "paho"; private static final long QUIESCE_TIMEOUT = 30000; // ms @@ -105,273 +116,337 @@ public class MqttAsyncClient implements IMqttAsyncClient { private MqttConnectOptions connOpts; private Object userContext; private Timer reconnectTimer; // Automatic reconnect timer - private static int reconnectDelay = 1000; // Reconnect delay, starts at 1 second + private static int reconnectDelay = 1000; // Reconnect delay, starts at 1 + // second private boolean reconnecting = false; private static Object clientLock = new Object(); // Simple lock - - - private ScheduledExecutorService executorService; /** - * Create an MqttAsyncClient that is used to communicate with an MQTT server. + * Create an MqttAsyncClient that is used to communicate with an MQTT + * server. *

- * The address of a server can be specified on the constructor. Alternatively - * a list containing one or more servers can be specified using the - * {@link MqttConnectOptions#setServerURIs(String[]) setServerURIs} method - * on MqttConnectOptions. + * The address of a server can be specified on the constructor. + * Alternatively a list containing one or more servers can be specified + * using the {@link MqttConnectOptions#setServerURIs(String[]) + * setServerURIs} method on MqttConnectOptions. * - *

The serverURI parameter is typically used with the - * the clientId parameter to form a key. The key - * is used to store and reference messages while they are being delivered. - * Hence the serverURI specified on the constructor must still be specified even if a list - * of servers is specified on an MqttConnectOptions object. - * The serverURI on the constructor must remain the same across - * restarts of the client for delivery of messages to be maintained from a given - * client to a given server or set of servers. + *

+ * The serverURI parameter is typically used with the the + * clientId parameter to form a key. The key is used to store + * and reference messages while they are being delivered. Hence the + * serverURI specified on the constructor must still be specified even if a + * list of servers is specified on an MqttConnectOptions object. The + * serverURI on the constructor must remain the same across restarts of the + * client for delivery of messages to be maintained from a given client to a + * given server or set of servers. * - *

The address of the server to connect to is specified as a URI. Two types of - * connection are supported tcp:// for a TCP connection and - * ssl:// for a TCP connection secured by SSL/TLS. - * For example:

+ *

+ * The address of the server to connect to is specified as a URI. Two types + * of connection are supported tcp:// for a TCP connection and + * ssl:// for a TCP connection secured by SSL/TLS. For example: + *

*
    - *
  • tcp://localhost:1883
    • - *
  • ssl://localhost:8883
    • + *
  • tcp://localhost:1883
    • + *
  • ssl://localhost:8883
    • *
*

- * If the port is not specified, it will - * default to 1883 for tcp://" URIs, and 8883 for ssl:// URIs. + * If the port is not specified, it will default to 1883 for + * tcp://" URIs, and 8883 for ssl:// URIs. *

* *

- * A client identifier clientId must be specified and be less that 65535 characters. - * It must be unique across all clients connecting to the same - * server. The clientId is used by the server to store data related to the client, - * hence it is important that the clientId remain the same when connecting to a server - * if durable subscriptions or reliable messaging are required. - *

A convenience method is provided to generate a random client id that - * should satisfy this criteria - {@link #generateClientId()}. As the client identifier - * is used by the server to identify a client when it reconnects, the client must use the - * same identifier between connections if durable subscriptions or reliable - * delivery of messages is required. + * A client identifier clientId must be specified and be less + * that 65535 characters. It must be unique across all clients connecting to + * the same server. The clientId is used by the server to store data related + * to the client, hence it is important that the clientId remain the same + * when connecting to a server if durable subscriptions or reliable + * messaging are required. + *

+ * A convenience method is provided to generate a random client id that + * should satisfy this criteria - {@link #generateClientId()}. As the client + * identifier is used by the server to identify a client when it reconnects, + * the client must use the same identifier between connections if durable + * subscriptions or reliable delivery of messages is required. *

*

* In Java SE, SSL can be configured in one of several ways, which the * client will use in the following order: *

*
    - *
  • Supplying an SSLSocketFactory - applications can - * use {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply - * a factory with the appropriate SSL settings.
    • - *
  • SSL Properties - applications can supply SSL settings as a - * simple Java Properties using {@link MqttConnectOptions#setSSLProperties(Properties)}.
    • - *
  • Use JVM settings - There are a number of standard - * Java system properties that can be used to configure key and trust stores.
    • + *
  • Supplying an SSLSocketFactory - + * applications can use + * {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply a + * factory with the appropriate SSL settings.
    • + *
  • SSL Properties - applications can supply SSL + * settings as a simple Java Properties using + * {@link MqttConnectOptions#setSSLProperties(Properties)}.
    • + *
  • Use JVM settings - There are a number of standard + * Java system properties that can be used to configure key and trust + * stores.
    • *
* - *

In Java ME, the platform settings are used for SSL connections.

+ *

+ * In Java ME, the platform settings are used for SSL connections. + *

* - *

An instance of the default persistence mechanism {@link MqttDefaultFilePersistence} - * is used by the client. To specify a different persistence mechanism or to turn - * off persistence, use the {@link #MqttAsyncClient(String, String, MqttClientPersistence)} + *

+ * An instance of the default persistence mechanism + * {@link MqttDefaultFilePersistence} is used by the client. To specify a + * different persistence mechanism or to turn off persistence, use the + * {@link #MqttAsyncClient(String, String, MqttClientPersistence)} * constructor. * - * @param serverURI the address of the server to connect to, specified as a URI. Can be overridden using - * {@link MqttConnectOptions#setServerURIs(String[])} - * @param clientId a client identifier that is unique on the server being connected to - * @throws IllegalArgumentException if the URI does not start with - * "tcp://", "ssl://" or "local://". - * @throws IllegalArgumentException if the clientId is null or is greater than 65535 characters in length - * @throws MqttException if any other problem was encountered + * @param serverURI + * the address of the server to connect to, specified as a URI. + * Can be overridden using + * {@link MqttConnectOptions#setServerURIs(String[])} + * @param clientId + * a client identifier that is unique on the server being + * connected to + * @throws IllegalArgumentException + * if the URI does not start with "tcp://", "ssl://" or + * "local://". + * @throws IllegalArgumentException + * if the clientId is null or is greater than 65535 characters + * in length + * @throws MqttException + * if any other problem was encountered */ public MqttAsyncClient(String serverURI, String clientId) throws MqttException { this(serverURI, clientId, new MqttDefaultFilePersistence()); } /** - * Create an MqttAsyncClient that is used to communicate with an MQTT server. + * Create an MqttAsyncClient that is used to communicate with an MQTT + * server. *

- * The address of a server can be specified on the constructor. Alternatively - * a list containing one or more servers can be specified using the - * {@link MqttConnectOptions#setServerURIs(String[]) setServerURIs} method - * on MqttConnectOptions. + * The address of a server can be specified on the constructor. + * Alternatively a list containing one or more servers can be specified + * using the {@link MqttConnectOptions#setServerURIs(String[]) + * setServerURIs} method on MqttConnectOptions. * - *

The serverURI parameter is typically used with the - * the clientId parameter to form a key. The key - * is used to store and reference messages while they are being delivered. - * Hence the serverURI specified on the constructor must still be specified even if a list - * of servers is specified on an MqttConnectOptions object. - * The serverURI on the constructor must remain the same across - * restarts of the client for delivery of messages to be maintained from a given - * client to a given server or set of servers. + *

+ * The serverURI parameter is typically used with the the + * clientId parameter to form a key. The key is used to store + * and reference messages while they are being delivered. Hence the + * serverURI specified on the constructor must still be specified even if a + * list of servers is specified on an MqttConnectOptions object. The + * serverURI on the constructor must remain the same across restarts of the + * client for delivery of messages to be maintained from a given client to a + * given server or set of servers. * - *

The address of the server to connect to is specified as a URI. Two types of - * connection are supported tcp:// for a TCP connection and - * ssl:// for a TCP connection secured by SSL/TLS. - * For example:

+ *

+ * The address of the server to connect to is specified as a URI. Two types + * of connection are supported tcp:// for a TCP connection and + * ssl:// for a TCP connection secured by SSL/TLS. For example: + *

*
    - *
  • tcp://localhost:1883
    • - *
  • ssl://localhost:8883
    • + *
  • tcp://localhost:1883
    • + *
  • ssl://localhost:8883
    • *
*

- * If the port is not specified, it will - * default to 1883 for tcp://" URIs, and 8883 for ssl:// URIs. + * If the port is not specified, it will default to 1883 for + * tcp://" URIs, and 8883 for ssl:// URIs. *

* *

- * A client identifier clientId must be specified and be less that 65535 characters. - * It must be unique across all clients connecting to the same - * server. The clientId is used by the server to store data related to the client, - * hence it is important that the clientId remain the same when connecting to a server - * if durable subscriptions or reliable messaging are required. - *

A convenience method is provided to generate a random client id that - * should satisfy this criteria - {@link #generateClientId()}. As the client identifier - * is used by the server to identify a client when it reconnects, the client must use the - * same identifier between connections if durable subscriptions or reliable - * delivery of messages is required. + * A client identifier clientId must be specified and be less + * that 65535 characters. It must be unique across all clients connecting to + * the same server. The clientId is used by the server to store data related + * to the client, hence it is important that the clientId remain the same + * when connecting to a server if durable subscriptions or reliable + * messaging are required. + *

+ * A convenience method is provided to generate a random client id that + * should satisfy this criteria - {@link #generateClientId()}. As the client + * identifier is used by the server to identify a client when it reconnects, + * the client must use the same identifier between connections if durable + * subscriptions or reliable delivery of messages is required. *

*

* In Java SE, SSL can be configured in one of several ways, which the * client will use in the following order: *

*
    - *
  • Supplying an SSLSocketFactory - applications can - * use {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply - * a factory with the appropriate SSL settings.
    • - *
  • SSL Properties - applications can supply SSL settings as a - * simple Java Properties using {@link MqttConnectOptions#setSSLProperties(Properties)}.
    • - *
  • Use JVM settings - There are a number of standard - * Java system properties that can be used to configure key and trust stores.
    • + *
  • Supplying an SSLSocketFactory - + * applications can use + * {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply a + * factory with the appropriate SSL settings.
    • + *
  • SSL Properties - applications can supply SSL + * settings as a simple Java Properties using + * {@link MqttConnectOptions#setSSLProperties(Properties)}.
    • + *
  • Use JVM settings - There are a number of standard + * Java system properties that can be used to configure key and trust + * stores.
    • *
* - *

In Java ME, the platform settings are used for SSL connections.

*

- * A persistence mechanism is used to enable reliable messaging. - * For messages sent at qualities of service (QoS) 1 or 2 to be reliably delivered, - * messages must be stored (on both the client and server) until the delivery of the message - * is complete. If messages are not safely stored when being delivered then - * a failure in the client or server can result in lost messages. A pluggable - * persistence mechanism is supported via the {@link MqttClientPersistence} - * interface. An implementer of this interface that safely stores messages - * must be specified in order for delivery of messages to be reliable. In - * addition {@link MqttConnectOptions#setCleanSession(boolean)} must be set - * to false. In the event that only QoS 0 messages are sent or received or + * In Java ME, the platform settings are used for SSL connections. + *

+ *

+ * A persistence mechanism is used to enable reliable messaging. For + * messages sent at qualities of service (QoS) 1 or 2 to be reliably + * delivered, messages must be stored (on both the client and server) until + * the delivery of the message is complete. If messages are not safely + * stored when being delivered then a failure in the client or server can + * result in lost messages. A pluggable persistence mechanism is supported + * via the {@link MqttClientPersistence} interface. An implementer of this + * interface that safely stores messages must be specified in order for + * delivery of messages to be reliable. In addition + * {@link MqttConnectOptions#setCleanSession(boolean)} must be set to false. + * In the event that only QoS 0 messages are sent or received or * cleanSession is set to true then a safe store is not needed. *

- *

An implementation of file-based persistence is provided in - * class {@link MqttDefaultFilePersistence} which will work in all Java SE based - * systems. If no persistence is needed, the persistence parameter - * can be explicitly set to null.

+ *

+ * An implementation of file-based persistence is provided in class + * {@link MqttDefaultFilePersistence} which will work in all Java SE based + * systems. If no persistence is needed, the persistence parameter can be + * explicitly set to null. + *

* - * @param serverURI the address of the server to connect to, specified as a URI. Can be overridden using - * {@link MqttConnectOptions#setServerURIs(String[])} - * @param clientId a client identifier that is unique on the server being connected to - * @param persistence the persistence class to use to store in-flight message. If null then the - * default persistence mechanism is used - * @throws MqttException if any other problem was encountered + * @param serverURI + * the address of the server to connect to, specified as a URI. + * Can be overridden using + * {@link MqttConnectOptions#setServerURIs(String[])} + * @param clientId + * a client identifier that is unique on the server being + * connected to + * @param persistence + * the persistence class to use to store in-flight message. If + * null then the default persistence mechanism is used + * @throws MqttException + * if any other problem was encountered */ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence) throws MqttException { this(serverURI, clientId, persistence, new TimerPingSender()); } - public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence, MqttPingSender pingSender) throws MqttException { + public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence, + MqttPingSender pingSender) throws MqttException { this(serverURI, clientId, persistence, pingSender, null); } /** - * Create an MqttAsyncClient that is used to communicate with an MQTT server. + * Create an MqttAsyncClient that is used to communicate with an MQTT + * server. *

- * The address of a server can be specified on the constructor. Alternatively - * a list containing one or more servers can be specified using the - * {@link MqttConnectOptions#setServerURIs(String[]) setServerURIs} method - * on MqttConnectOptions. + * The address of a server can be specified on the constructor. + * Alternatively a list containing one or more servers can be specified + * using the {@link MqttConnectOptions#setServerURIs(String[]) + * setServerURIs} method on MqttConnectOptions. * - *

The serverURI parameter is typically used with the - * the clientId parameter to form a key. The key - * is used to store and reference messages while they are being delivered. - * Hence the serverURI specified on the constructor must still be specified even if a list - * of servers is specified on an MqttConnectOptions object. - * The serverURI on the constructor must remain the same across - * restarts of the client for delivery of messages to be maintained from a given - * client to a given server or set of servers. + *

+ * The serverURI parameter is typically used with the the + * clientId parameter to form a key. The key is used to store + * and reference messages while they are being delivered. Hence the + * serverURI specified on the constructor must still be specified even if a + * list of servers is specified on an MqttConnectOptions object. The + * serverURI on the constructor must remain the same across restarts of the + * client for delivery of messages to be maintained from a given client to a + * given server or set of servers. * - *

The address of the server to connect to is specified as a URI. Two types of - * connection are supported tcp:// for a TCP connection and - * ssl:// for a TCP connection secured by SSL/TLS. - * For example:

+ *

+ * The address of the server to connect to is specified as a URI. Two types + * of connection are supported tcp:// for a TCP connection and + * ssl:// for a TCP connection secured by SSL/TLS. For example: + *

*
    - *
  • tcp://localhost:1883
    • - *
  • ssl://localhost:8883
    • + *
  • tcp://localhost:1883
    • + *
  • ssl://localhost:8883
    • *
*

- * If the port is not specified, it will - * default to 1883 for tcp://" URIs, and 8883 for ssl:// URIs. + * If the port is not specified, it will default to 1883 for + * tcp://" URIs, and 8883 for ssl:// URIs. *

* *

- * A client identifier clientId must be specified and be less that 65535 characters. - * It must be unique across all clients connecting to the same - * server. The clientId is used by the server to store data related to the client, - * hence it is important that the clientId remain the same when connecting to a server - * if durable subscriptions or reliable messaging are required. - *

A convenience method is provided to generate a random client id that - * should satisfy this criteria - {@link #generateClientId()}. As the client identifier - * is used by the server to identify a client when it reconnects, the client must use the - * same identifier between connections if durable subscriptions or reliable - * delivery of messages is required. + * A client identifier clientId must be specified and be less + * that 65535 characters. It must be unique across all clients connecting to + * the same server. The clientId is used by the server to store data related + * to the client, hence it is important that the clientId remain the same + * when connecting to a server if durable subscriptions or reliable + * messaging are required. + *

+ * A convenience method is provided to generate a random client id that + * should satisfy this criteria - {@link #generateClientId()}. As the client + * identifier is used by the server to identify a client when it reconnects, + * the client must use the same identifier between connections if durable + * subscriptions or reliable delivery of messages is required. *

*

* In Java SE, SSL can be configured in one of several ways, which the * client will use in the following order: *

*
    - *
  • Supplying an SSLSocketFactory - applications can - * use {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply - * a factory with the appropriate SSL settings.
    • - *
  • SSL Properties - applications can supply SSL settings as a - * simple Java Properties using {@link MqttConnectOptions#setSSLProperties(Properties)}.
    • - *
  • Use JVM settings - There are a number of standard - * Java system properties that can be used to configure key and trust stores.
    • + *
  • Supplying an SSLSocketFactory - + * applications can use + * {@link MqttConnectOptions#setSocketFactory(SocketFactory)} to supply a + * factory with the appropriate SSL settings.
    • + *
  • SSL Properties - applications can supply SSL + * settings as a simple Java Properties using + * {@link MqttConnectOptions#setSSLProperties(Properties)}.
    • + *
  • Use JVM settings - There are a number of standard + * Java system properties that can be used to configure key and trust + * stores.
    • *
* - *

In Java ME, the platform settings are used for SSL connections.

*

- * A persistence mechanism is used to enable reliable messaging. - * For messages sent at qualities of service (QoS) 1 or 2 to be reliably delivered, - * messages must be stored (on both the client and server) until the delivery of the message - * is complete. If messages are not safely stored when being delivered then - * a failure in the client or server can result in lost messages. A pluggable - * persistence mechanism is supported via the {@link MqttClientPersistence} - * interface. An implementer of this interface that safely stores messages - * must be specified in order for delivery of messages to be reliable. In - * addition {@link MqttConnectOptions#setCleanSession(boolean)} must be set - * to false. In the event that only QoS 0 messages are sent or received or + * In Java ME, the platform settings are used for SSL connections. + *

+ *

+ * A persistence mechanism is used to enable reliable messaging. For + * messages sent at qualities of service (QoS) 1 or 2 to be reliably + * delivered, messages must be stored (on both the client and server) until + * the delivery of the message is complete. If messages are not safely + * stored when being delivered then a failure in the client or server can + * result in lost messages. A pluggable persistence mechanism is supported + * via the {@link MqttClientPersistence} interface. An implementer of this + * interface that safely stores messages must be specified in order for + * delivery of messages to be reliable. In addition + * {@link MqttConnectOptions#setCleanSession(boolean)} must be set to false. + * In the event that only QoS 0 messages are sent or received or * cleanSession is set to true then a safe store is not needed. *

- *

An implementation of file-based persistence is provided in - * class {@link MqttDefaultFilePersistence} which will work in all Java SE based - * systems. If no persistence is needed, the persistence parameter - * can be explicitly set to null.

+ *

+ * An implementation of file-based persistence is provided in class + * {@link MqttDefaultFilePersistence} which will work in all Java SE based + * systems. If no persistence is needed, the persistence parameter can be + * explicitly set to null. + *

* - * @param serverURI the address of the server to connect to, specified as a URI. Can be overridden using - * {@link MqttConnectOptions#setServerURIs(String[])} - * @param clientId a client identifier that is unique on the server being connected to - * @param persistence the persistence class to use to store in-flight message. If null then the - * default persistence mechanism is used - * @param pingSender Custom {@link MqttPingSender} implementation. - * @param executorService used for managing threads. If null then a newFixedThreadPool is used. - * @throws IllegalArgumentException if the URI does not start with - * "tcp://", "ssl://" or "local://" - * @throws IllegalArgumentException if the clientId is null or is greater than 65535 characters in length - * @throws MqttException if any other problem was encountered + * @param serverURI + * the address of the server to connect to, specified as a URI. + * Can be overridden using + * {@link MqttConnectOptions#setServerURIs(String[])} + * @param clientId + * a client identifier that is unique on the server being + * connected to + * @param persistence + * the persistence class to use to store in-flight message. If + * null then the default persistence mechanism is used + * @param pingSender + * Custom {@link MqttPingSender} implementation. + * @param executorService + * used for managing threads. If null then a newFixedThreadPool + * is used. + * @throws IllegalArgumentException + * if the URI does not start with "tcp://", "ssl://" or + * "local://" + * @throws IllegalArgumentException + * if the clientId is null or is greater than 65535 characters + * in length + * @throws MqttException + * if any other problem was encountered */ - public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence, MqttPingSender pingSender, ScheduledExecutorService executorService) throws MqttException { - final String methodName = "MqttAsyncClient"; + public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence persistence, + MqttPingSender pingSender, ScheduledExecutorService executorService) throws MqttException { + final String methodName = "MqttAsyncClient"; log.setResourceName(clientId); - if (clientId == null) { //Support empty client Id, 3.1.1 standard + if (clientId == null) { // Support empty client Id, 3.1.1 standard throw new IllegalArgumentException("Null clientId"); } // Count characters, surrogate pairs count as one character. @@ -381,7 +456,7 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence i++; clientIdLength++; } - if ( clientIdLength > 65535) { + if (clientIdLength > 65535) { throw new IllegalArgumentException("ClientId longer than 65535 characters"); } @@ -401,7 +476,7 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence } // @TRACE 101= ClientID={0} ServerURI={1} PersistenceType={2} - log.fine(CLASS_NAME,methodName,"101",new Object[]{clientId,serverURI,persistence}); + log.fine(CLASS_NAME, methodName, "101", new Object[] { clientId, serverURI, persistence }); this.persistence.open(clientId, serverURI); this.comms = new ClientComms(this, this.persistence, pingSender, this.executorService); @@ -410,38 +485,42 @@ public MqttAsyncClient(String serverURI, String clientId, MqttClientPersistence } - - /** - * @param ch the character to check. + * @param ch + * the character to check. * @return returns 'true' if the character is a high-surrogate code unit */ protected static boolean Character_isHighSurrogate(char ch) { - return(ch >= MIN_HIGH_SURROGATE) && (ch <= MAX_HIGH_SURROGATE); + return (ch >= MIN_HIGH_SURROGATE) && (ch <= MAX_HIGH_SURROGATE); } /** - * Factory method to create an array of network modules, one for - * each of the supplied URIs + * Factory method to create an array of network modules, one for each of the + * supplied URIs * - * @param address the URI for the server. - * @param options the {@link MqttConnectOptions} for the connection. + * @param address + * the URI for the server. + * @param options + * the {@link MqttConnectOptions} for the connection. * @return a network module appropriate to the specified address. - * @throws MqttException if an exception occurs creating the network Modules - * @throws MqttSecurityException if an issue occurs creating an SSL / TLS Socket + * @throws MqttException + * if an exception occurs creating the network Modules + * @throws MqttSecurityException + * if an issue occurs creating an SSL / TLS Socket */ - protected NetworkModule[] createNetworkModules(String address, MqttConnectOptions options) throws MqttException, MqttSecurityException { + protected NetworkModule[] createNetworkModules(String address, MqttConnectOptions options) + throws MqttException, MqttSecurityException { final String methodName = "createNetworkModules"; // @TRACE 116=URI={0} - log.fine(CLASS_NAME, methodName, "116", new Object[]{address}); + log.fine(CLASS_NAME, methodName, "116", new Object[] { address }); NetworkModule[] networkModules = null; String[] serverURIs = options.getServerURIs(); String[] array = null; if (serverURIs == null) { - array = new String[]{address}; + array = new String[] { address }; } else if (serverURIs.length == 0) { - array = new String[]{address}; + array = new String[] { address }; } else { array = serverURIs; } @@ -476,6 +555,21 @@ private NetworkModule createNetworkModule(String address, MqttConnectOptions opt URI uri; try { uri = new URI(address); + // If the returned uri contains no host and the address contains underscores, + // then it's likely that Java did not parse the URI + if(uri.getHost() == null && address.contains("_")){ + try { + final Field hostField = URI.class.getDeclaredField("host"); + hostField.setAccessible(true); + // Get everything after the scheme:// + String shortAddress = address.substring(uri.getScheme().length() + 3); + hostField.set(uri, getHostName(shortAddress)); + + } catch (NoSuchFieldException | SecurityException | IllegalArgumentException | IllegalAccessException e) { + throw ExceptionHelper.createMqttException(e.getCause()); + } + + } } catch (URISyntaxException e) { throw new IllegalArgumentException("Malformed URI: " + address + ", " + e.getMessage()); } @@ -579,30 +673,56 @@ else if ((factory instanceof SSLSocketFactory) == false) { return netModule; } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + private String getHostName(String uri) { + int portIndex = uri.indexOf(':'); + if (portIndex == -1) { + portIndex = uri.indexOf('/'); + } + if (portIndex == -1) { + portIndex = uri.length(); + } + return uri.substring(0, portIndex); + } + + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ public IMqttToken connect(Object userContext, IMqttActionListener callback) throws MqttException, MqttSecurityException { return this.connect(new MqttConnectOptions(), userContext, callback); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect() */ public IMqttToken connect() throws MqttException, MqttSecurityException { return this.connect(null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(org.eclipse.paho.client.mqttv3.MqttConnectOptions) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(org.eclipse.paho. + * client.mqttv3.MqttConnectOptions) */ public IMqttToken connect(MqttConnectOptions options) throws MqttException, MqttSecurityException { - return this.connect(options, null,null); + return this.connect(options, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(org.eclipse.paho.client.mqttv3.MqttConnectOptions, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#connect(org.eclipse.paho. + * client.mqttv3.MqttConnectOptions, java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ public IMqttToken connect(MqttConnectOptions options, Object userContext, IMqttActionListener callback) throws MqttException, MqttSecurityException { @@ -619,37 +739,35 @@ public IMqttToken connect(MqttConnectOptions options, Object userContext, IMqttA if (comms.isClosed()) { throw new MqttException(MqttException.REASON_CODE_CLIENT_CLOSED); } - if(options == null){ + if (options == null) { options = new MqttConnectOptions(); } this.connOpts = options; this.userContext = userContext; final boolean automaticReconnect = options.isAutomaticReconnect(); - // @TRACE 103=cleanSession={0} connectionTimeout={1} TimekeepAlive={2} userName={3} password={4} will={5} userContext={6} callback={7} - log.fine(CLASS_NAME,methodName, "103", - new Object[]{ - Boolean.valueOf(options.isCleanSession()), - new Integer(options.getConnectionTimeout()), - new Integer(options.getKeepAliveInterval()), - options.getUserName(), - ((null == options.getPassword())?"[null]":"[notnull]"), - ((null == options.getWillMessage())?"[null]":"[notnull]"), - userContext, - callback }); + // @TRACE 103=cleanSession={0} connectionTimeout={1} TimekeepAlive={2} + // userName={3} password={4} will={5} userContext={6} callback={7} + log.fine(CLASS_NAME, methodName, "103", + new Object[] { Boolean.valueOf(options.isCleanSession()), new Integer(options.getConnectionTimeout()), + new Integer(options.getKeepAliveInterval()), options.getUserName(), + ((null == options.getPassword()) ? "[null]" : "[notnull]"), + ((null == options.getWillMessage()) ? "[null]" : "[notnull]"), userContext, callback }); comms.setNetworkModules(createNetworkModules(serverURI, options)); comms.setReconnectCallback(new MqttReconnectCallback(automaticReconnect)); - - // Insert our own callback to iterate through the URIs till the connect succeeds + // Insert our own callback to iterate through the URIs till the connect + // succeeds MqttToken userToken = new MqttToken(getClientId()); - ConnectActionListener connectActionListener = new ConnectActionListener(this, persistence, comms, options, userToken, userContext, callback, reconnecting); + ConnectActionListener connectActionListener = new ConnectActionListener(this, persistence, comms, options, + userToken, userContext, callback, reconnecting); userToken.setActionCallback(connectActionListener); userToken.setUserContext(this); - // If we are using the MqttCallbackExtended, set it on the connectActionListener - if(this.mqttCallback instanceof MqttCallbackExtended){ - connectActionListener.setMqttCallbackExtended((MqttCallbackExtended)this.mqttCallback); + // If we are using the MqttCallbackExtended, set it on the + // connectActionListener + if (this.mqttCallback instanceof MqttCallbackExtended) { + connectActionListener.setMqttCallbackExtended((MqttCallbackExtended) this.mqttCallback); } comms.setNetworkModuleIndex(0); @@ -657,36 +775,47 @@ public IMqttToken connect(MqttConnectOptions options, Object userContext, IMqttA return userToken; } - - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnect(java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnect(java.lang. + * Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken disconnect( Object userContext, IMqttActionListener callback) throws MqttException { + public IMqttToken disconnect(Object userContext, IMqttActionListener callback) throws MqttException { return this.disconnect(QUIESCE_TIMEOUT, userContext, callback); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnect() */ public IMqttToken disconnect() throws MqttException { return this.disconnect(null, null); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnect(long) */ public IMqttToken disconnect(long quiesceTimeout) throws MqttException { return this.disconnect(quiesceTimeout, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnect(long, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnect(long, + * java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken disconnect(long quiesceTimeout, Object userContext, IMqttActionListener callback) throws MqttException { + public IMqttToken disconnect(long quiesceTimeout, Object userContext, IMqttActionListener callback) + throws MqttException { final String methodName = "disconnect"; // @TRACE 104=> quiesceTimeout={0} userContext={1} callback={2} - log.fine(CLASS_NAME,methodName, "104",new Object[]{ new Long(quiesceTimeout), userContext, callback}); + log.fine(CLASS_NAME, methodName, "104", new Object[] { new Long(quiesceTimeout), userContext, callback }); MqttToken token = new MqttToken(getClientId()); token.setActionCallback(callback); @@ -696,18 +825,19 @@ public IMqttToken disconnect(long quiesceTimeout, Object userContext, IMqttActio try { comms.disconnect(disconnect, quiesceTimeout, token); } catch (MqttException ex) { - //@TRACE 105=< exception - log.fine(CLASS_NAME,methodName,"105",null,ex); + // @TRACE 105=< exception + log.fine(CLASS_NAME, methodName, "105", null, ex); throw ex; } - //@TRACE 108=< - log.fine(CLASS_NAME,methodName,"108"); + // @TRACE 108=< + log.fine(CLASS_NAME, methodName, "108"); return token; } /* * (non-Javadoc) + * * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnectForcibly() */ public void disconnectForcibly() throws MqttException { @@ -716,7 +846,9 @@ public void disconnectForcibly() throws MqttException { /* * (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnectForcibly(long) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnectForcibly(long) */ public void disconnectForcibly(long disconnectTimeout) throws MqttException { disconnectForcibly(QUIESCE_TIMEOUT, disconnectTimeout); @@ -724,43 +856,60 @@ public void disconnectForcibly(long disconnectTimeout) throws MqttException { /* * (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnectForcibly(long, long) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#disconnectForcibly(long, + * long) */ - public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout) throws MqttException{ + public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout) throws MqttException { comms.disconnectForcibly(quiesceTimeout, disconnectTimeout); } /** - * Disconnects from the server forcibly to reset all the states. Could be useful when disconnect attempt failed. + * Disconnects from the server forcibly to reset all the states. Could be + * useful when disconnect attempt failed. *

- * Because the client is able to establish the TCP/IP connection to a none MQTT server and it will certainly fail to - * send the disconnect packet. + * Because the client is able to establish the TCP/IP connection to a none + * MQTT server and it will certainly fail to send the disconnect packet. * - * @param quiesceTimeout the amount of time in milliseconds to allow for existing work to finish before - * disconnecting. A value of zero or less means the client will not quiesce. - * @param disconnectTimeout the amount of time in milliseconds to allow send disconnect packet to server. - * @param sendDisconnectPacket if true, will send the disconnect packet to the server - * @throws MqttException if any unexpected error + * @param quiesceTimeout + * the amount of time in milliseconds to allow for existing work + * to finish before disconnecting. A value of zero or less means + * the client will not quiesce. + * @param disconnectTimeout + * the amount of time in milliseconds to allow send disconnect + * packet to server. + * @param sendDisconnectPacket + * if true, will send the disconnect packet to the server + * @throws MqttException + * if any unexpected error */ - public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout, boolean sendDisconnectPacket) throws MqttException { - comms.disconnectForcibly(quiesceTimeout, disconnectTimeout, sendDisconnectPacket); - } + public void disconnectForcibly(long quiesceTimeout, long disconnectTimeout, boolean sendDisconnectPacket) + throws MqttException { + comms.disconnectForcibly(quiesceTimeout, disconnectTimeout, sendDisconnectPacket); + } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see IMqttAsyncClient#isConnected() */ public boolean isConnected() { return comms.isConnected(); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see IMqttAsyncClient#getClientId() */ public String getClientId() { return clientId; } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see IMqttAsyncClient#getServerURI() */ public String getServerURI() { @@ -768,118 +917,143 @@ public String getServerURI() { } /** - * Returns the currently connected Server URI - * Implemented due to: https://bugs.eclipse.org/bugs/show_bug.cgi?id=481097 + * Returns the currently connected Server URI Implemented due to: + * https://bugs.eclipse.org/bugs/show_bug.cgi?id=481097 * * Where getServerURI only returns the URI that was provided in * MqttAsyncClient's constructor, getCurrentServerURI returns the URI of the - * Server that the client is currently connected to. This would be different in scenarios - * where multiple server URIs have been provided to the MqttConnectOptions. + * Server that the client is currently connected to. This would be different + * in scenarios where multiple server URIs have been provided to the + * MqttConnectOptions. * * @return the currently connected server URI */ - public String getCurrentServerURI(){ + public String getCurrentServerURI() { return comms.getNetworkModules()[comms.getNetworkModuleIndex()].getServerURI(); } /** * Get a topic object which can be used to publish messages. - *

There are two alternative methods that should be used in preference to this one when publishing a message:

+ *

+ * There are two alternative methods that should be used in preference to + * this one when publishing a message: + *

*
    - *
  • {@link MqttAsyncClient#publish(String, MqttMessage)} to publish a message in a non-blocking manner or
    • - *
  • {@link MqttClient#publish(String, MqttMessage)} to publish a message in a blocking manner
    • + *
  • {@link MqttAsyncClient#publish(String, MqttMessage)} to publish a + * message in a non-blocking manner or
    • + *
  • {@link MqttClient#publish(String, MqttMessage)} to publish a message + * in a blocking manner
    • *
- *

When you build an application, - * the design of the topic tree should take into account the following principles - * of topic name syntax and semantics:

+ *

+ * When you build an application, the design of the topic tree should take + * into account the following principles of topic name syntax and semantics: + *

* *
    - *
  • A topic must be at least one character long.
    • - *
  • Topic names are case sensitive. For example, ACCOUNTS and Accounts are - * two different topics.
    • - *
  • Topic names can include the space character. For example, Accounts - * payable is a valid topic.
    • - *
  • A leading "/" creates a distinct topic. For example, /finance is - * different from finance. /finance matches "+/+" and "/+", but - * not "+".
    • - *
  • Do not include the null character (Unicode \x0000) in - * any topic.
    • + *
  • A topic must be at least one character long.
    • + *
  • Topic names are case sensitive. For example, ACCOUNTS and + * Accounts are two different topics.
    • + *
  • Topic names can include the space character. For example, + * Accounts payable is a valid topic.
    • + *
  • A leading "/" creates a distinct topic. For example, + * /finance is different from finance. /finance + * matches "+/+" and "/+", but not "+".
    • + *
  • Do not include the null character (Unicode \x0000) in any topic.
    • *
* - *

The following principles apply to the construction and content of a topic - * tree:

+ *

+ * The following principles apply to the construction and content of a topic + * tree: + *

* *
    - *
  • The length is limited to 64k but within that there are no limits to the - * number of levels in a topic tree.
    • - *
  • There can be any number of root nodes; that is, there can be any number - * of topic trees.
    • - *
+ *
  • The length is limited to 64k but within that there are no limits to + * the number of levels in a topic tree.
    • + *
  • There can be any number of root nodes; that is, there can be any + * number of topic trees.
    • + * * - * @param topic the topic to use, for example "finance/stock/ibm". - * @return an MqttTopic object, which can be used to publish messages to - * the topic. - * @throws IllegalArgumentException if the topic contains a '+' or '#' - * wildcard character. + * @param topic + * the topic to use, for example "finance/stock/ibm". + * @return an MqttTopic object, which can be used to publish messages to the + * topic. + * @throws IllegalArgumentException + * if the topic contains a '+' or '#' wildcard character. */ protected MqttTopic getTopic(String topic) { - MqttTopic.validate(topic, false/*wildcards NOT allowed*/); + MqttTopic.validate(topic, false/* wildcards NOT allowed */); - MqttTopic result = (MqttTopic)topics.get(topic); + MqttTopic result = (MqttTopic) topics.get(topic); if (result == null) { result = new MqttTopic(topic, comms); - topics.put(topic,result); + topics.put(topic, result); } return result; } - /* (non-Javadoc) - * Check and send a ping if needed. - *

    By default, client sends PingReq to server to keep the connection to - * server. For some platforms which cannot use this mechanism, such as Android, - * developer needs to handle the ping request manually with this method. - *

    + /* + * (non-Javadoc) Check and send a ping if needed.

    By default, client + * sends PingReq to server to keep the connection to server. For some + * platforms which cannot use this mechanism, such as Android, developer + * needs to handle the ping request manually with this method.

    * - * @throws MqttException for other errors encountered while publishing the message. + * @throws MqttException for other errors encountered while publishing the + * message. */ - public IMqttToken checkPing(Object userContext, IMqttActionListener callback) throws MqttException{ + public IMqttToken checkPing(Object userContext, IMqttActionListener callback) throws MqttException { final String methodName = "ping"; MqttToken token; - //@TRACE 117=> - log.fine(CLASS_NAME,methodName,"117"); + // @TRACE 117=> + log.fine(CLASS_NAME, methodName, "117"); token = comms.checkForActivity(); - //@TRACE 118=< - log.fine(CLASS_NAME,methodName,"118"); + // @TRACE 118=< + log.fine(CLASS_NAME, methodName, "118"); return token; } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String, int, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String, int, java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken subscribe(String topicFilter, int qos, Object userContext, IMqttActionListener callback) throws MqttException { - return this.subscribe(new String[] {topicFilter}, new int[] {qos}, userContext, callback); + public IMqttToken subscribe(String topicFilter, int qos, Object userContext, IMqttActionListener callback) + throws MqttException { + return this.subscribe(new String[] { topicFilter }, new int[] { qos }, userContext, callback); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String, int) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String, int) */ public IMqttToken subscribe(String topicFilter, int qos) throws MqttException { - return this.subscribe(new String[] {topicFilter}, new int[] {qos}, null, null); + return this.subscribe(new String[] { topicFilter }, new int[] { qos }, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String[], int[]) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String[], int[]) */ public IMqttToken subscribe(String[] topicFilters, int[] qos) throws MqttException { return this.subscribe(topicFilters, qos, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String[], int[], java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String[], int[], java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback) throws MqttException { + public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback) + throws MqttException { final String methodName = "subscribe"; if (topicFilters.length != qos.length) { @@ -892,19 +1066,19 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext } // Only Generate Log string if we are logging at FINE level - if(log.isLoggable(Logger.FINE)){ + if (log.isLoggable(Logger.FINE)) { StringBuffer subs = new StringBuffer(); - for (int i=0;i0) { + for (int i = 0; i < topicFilters.length; i++) { + if (i > 0) { subs.append(", "); } subs.append("topic=").append(topicFilters[i]).append(" qos=").append(qos[i]); - //Check if the topic filter is valid before subscribing - MqttTopic.validate(topicFilters[i], true/*allow wildcards*/); + // Check if the topic filter is valid before subscribing + MqttTopic.validate(topicFilters[i], true/* allow wildcards */); } - //@TRACE 106=Subscribe topicFilter={0} userContext={1} callback={2} - log.fine(CLASS_NAME,methodName,"106",new Object[]{subs.toString(), userContext, callback}); + // @TRACE 106=Subscribe topicFilter={0} userContext={1} callback={2} + log.fine(CLASS_NAME, methodName, "106", new Object[] { subs.toString(), userContext, callback }); } MqttToken token = new MqttToken(getClientId()); @@ -915,35 +1089,51 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext MqttSubscribe register = new MqttSubscribe(topicFilters, qos); comms.sendNoWait(register, token); - //@TRACE 109=< - log.fine(CLASS_NAME,methodName,"109"); + // @TRACE 109=< + log.fine(CLASS_NAME, methodName, "109"); return token; } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String, int, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String, int, java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken subscribe(String topicFilter, int qos, Object userContext, IMqttActionListener callback, IMqttMessageListener messageListener) throws MqttException { + public IMqttToken subscribe(String topicFilter, int qos, Object userContext, IMqttActionListener callback, + IMqttMessageListener messageListener) throws MqttException { - return this.subscribe(new String[] {topicFilter}, new int[] {qos}, userContext, callback, new IMqttMessageListener[] {messageListener}); + return this.subscribe(new String[] { topicFilter }, new int[] { qos }, userContext, callback, + new IMqttMessageListener[] { messageListener }); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String, int) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String, int) */ - public IMqttToken subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) throws MqttException { - return this.subscribe(new String[] {topicFilter}, new int[] {qos}, null, null, new IMqttMessageListener[] {messageListener}); + public IMqttToken subscribe(String topicFilter, int qos, IMqttMessageListener messageListener) + throws MqttException { + return this.subscribe(new String[] { topicFilter }, new int[] { qos }, null, null, + new IMqttMessageListener[] { messageListener }); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang.String[], int[]) + /* + * (non-Javadoc) + * + * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#subscribe(java.lang. + * String[], int[]) */ - public IMqttToken subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) throws MqttException { + public IMqttToken subscribe(String[] topicFilters, int[] qos, IMqttMessageListener[] messageListeners) + throws MqttException { return this.subscribe(topicFilters, qos, null, null, messageListeners); } - public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback, IMqttMessageListener[] messageListeners) throws MqttException { + public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext, IMqttActionListener callback, + IMqttMessageListener[] messageListeners) throws MqttException { if ((messageListeners.length != qos.length) || (qos.length != topicFilters.length)) { throw new IllegalArgumentException(); @@ -959,57 +1149,75 @@ public IMqttToken subscribe(String[] topicFilters, int[] qos, Object userContext return token; } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang.String, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang. + * String, java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken unsubscribe(String topicFilter, Object userContext, IMqttActionListener callback) throws MqttException { - return unsubscribe(new String[] {topicFilter}, userContext, callback); + public IMqttToken unsubscribe(String topicFilter, Object userContext, IMqttActionListener callback) + throws MqttException { + return unsubscribe(new String[] { topicFilter }, userContext, callback); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang.String) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang. + * String) */ public IMqttToken unsubscribe(String topicFilter) throws MqttException { - return unsubscribe(new String[] {topicFilter}, null, null); + return unsubscribe(new String[] { topicFilter }, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang.String[]) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang. + * String[]) */ public IMqttToken unsubscribe(String[] topicFilters) throws MqttException { return unsubscribe(topicFilters, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang.String[], java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#unsubscribe(java.lang. + * String[], java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttToken unsubscribe(String[] topicFilters, Object userContext, IMqttActionListener callback) throws MqttException { + public IMqttToken unsubscribe(String[] topicFilters, Object userContext, IMqttActionListener callback) + throws MqttException { final String methodName = "unsubscribe"; // Only Generate Log string if we are logging at FINE level - if(log.isLoggable(Logger.FINE)){ + if (log.isLoggable(Logger.FINE)) { String subs = ""; - for (int i=0;i0) { - subs+=", "; + for (int i = 0; i < topicFilters.length; i++) { + if (i > 0) { + subs += ", "; } - subs+=topicFilters[i]; + subs += topicFilters[i]; } - //@TRACE 107=Unsubscribe topic={0} userContext={1} callback={2} - log.fine(CLASS_NAME, methodName,"107",new Object[]{subs, userContext, callback}); + // @TRACE 107=Unsubscribe topic={0} userContext={1} callback={2} + log.fine(CLASS_NAME, methodName, "107", new Object[] { subs, userContext, callback }); } - for (int i=0;iWhen cleanSession is set to false, an application must ensure it uses the - * same client identifier when it reconnects to the server to resume state and maintain - * assured message delivery.

    + * Returns a randomly generated client identifier based on the the fixed + * prefix (paho) and the system time. + *

    + * When cleanSession is set to false, an application must ensure it uses the + * same client identifier when it reconnects to the server to resume state + * and maintain assured message delivery. + *

    + * * @return a generated client identifier * @see MqttConnectOptions#setCleanSession(boolean) */ public static String generateClientId() { - //length of nanoTime = 15, so total length = 19 < 65535(defined in spec) + // length of nanoTime = 15, so total length = 19 < 65535(defined in + // spec) return CLIENT_ID_PREFIX + System.nanoTime(); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see IMqttAsyncClient#getPendingDeliveryTokens() */ public IMqttDeliveryToken[] getPendingDeliveryTokens() { return comms.getPendingDeliveryTokens(); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, byte[], int, boolean, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, + * byte[], int, boolean, java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttDeliveryToken publish(String topic, byte[] payload, int qos, - boolean retained, Object userContext, IMqttActionListener callback) throws MqttException, - MqttPersistenceException { + public IMqttDeliveryToken publish(String topic, byte[] payload, int qos, boolean retained, Object userContext, + IMqttActionListener callback) throws MqttException, MqttPersistenceException { MqttMessage message = new MqttMessage(payload); message.setQos(qos); message.setRetained(retained); return this.publish(topic, message, userContext, callback); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, byte[], int, boolean) + + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, + * byte[], int, boolean) */ - public IMqttDeliveryToken publish(String topic, byte[] payload, int qos, - boolean retained) throws MqttException, MqttPersistenceException { + public IMqttDeliveryToken publish(String topic, byte[] payload, int qos, boolean retained) + throws MqttException, MqttPersistenceException { return this.publish(topic, payload, qos, retained, null, null); } - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, org.eclipse.paho.client.mqttv3.MqttMessage) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, + * org.eclipse.paho.client.mqttv3.MqttMessage) */ - public IMqttDeliveryToken publish(String topic, MqttMessage message) throws MqttException, MqttPersistenceException { + public IMqttDeliveryToken publish(String topic, MqttMessage message) + throws MqttException, MqttPersistenceException { return this.publish(topic, message, null, null); } - - /* (non-Javadoc) - * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, org.eclipse.paho.client.mqttv3.MqttMessage, java.lang.Object, org.eclipse.paho.client.mqttv3.IMqttActionListener) + /* + * (non-Javadoc) + * + * @see + * org.eclipse.paho.client.mqttv3.IMqttAsyncClient#publish(java.lang.String, + * org.eclipse.paho.client.mqttv3.MqttMessage, java.lang.Object, + * org.eclipse.paho.client.mqttv3.IMqttActionListener) */ - public IMqttDeliveryToken publish(String topic, MqttMessage message, Object userContext, IMqttActionListener callback) throws MqttException, - MqttPersistenceException { + public IMqttDeliveryToken publish(String topic, MqttMessage message, Object userContext, + IMqttActionListener callback) throws MqttException, MqttPersistenceException { final String methodName = "publish"; - //@TRACE 111=< topic={0} message={1}userContext={1} callback={2} - log.fine(CLASS_NAME,methodName,"111", new Object[] {topic, userContext, callback}); + // @TRACE 111=< topic={0} message={1}userContext={1} callback={2} + log.fine(CLASS_NAME, methodName, "111", new Object[] { topic, userContext, callback }); - //Checks if a topic is valid when publishing a message. - MqttTopic.validate(topic, false/*wildcards NOT allowed*/); + // Checks if a topic is valid when publishing a message. + MqttTopic.validate(topic, false/* wildcards NOT allowed */); MqttDeliveryToken token = new MqttDeliveryToken(getClientId()); token.setActionCallback(callback); token.setUserContext(userContext); token.setMessage(message); - token.internalTok.setTopics(new String[] {topic}); + token.internalTok.setTopics(new String[] { topic }); MqttPublish pubMsg = new MqttPublish(topic, message); comms.sendNoWait(pubMsg, token); - //@TRACE 112=< - log.fine(CLASS_NAME,methodName,"112"); + // @TRACE 112=< + log.fine(CLASS_NAME, methodName, "112"); return token; } /** * User triggered attempt to reconnect - * @throws MqttException if there is an issue with reconnecting + * + * @throws MqttException + * if there is an issue with reconnecting */ public void reconnect() throws MqttException { final String methodName = "reconnect"; - //@Trace 500=Attempting to reconnect client: {0} - log.fine(CLASS_NAME, methodName, "500", new Object[]{this.clientId}); - // Some checks to make sure that we're not attempting to reconnect an already connected client + // @Trace 500=Attempting to reconnect client: {0} + log.fine(CLASS_NAME, methodName, "500", new Object[] { this.clientId }); + // Some checks to make sure that we're not attempting to reconnect an + // already connected client if (comms.isConnected()) { throw ExceptionHelper.createMqttException(MqttException.REASON_CODE_CLIENT_CONNECTED); } @@ -1151,50 +1390,46 @@ public void reconnect() throws MqttException { attemptReconnect(); } - /** - * Attempts to reconnect the client to the server. - * If successful it will make sure that there are no further - * reconnects scheduled. However if the connect fails, the delay will double - * up to 128 seconds and will re-schedule the reconnect for after the delay. + * Attempts to reconnect the client to the server. If successful it will + * make sure that there are no further reconnects scheduled. However if the + * connect fails, the delay will double up to 128 seconds and will + * re-schedule the reconnect for after the delay. * * Any thrown exceptions are logged but not acted upon as it is assumed that * they are being thrown due to the server being offline and so reconnect * attempts will continue. */ - private void attemptReconnect(){ + private void attemptReconnect() { final String methodName = "attemptReconnect"; - //@Trace 500=Attempting to reconnect client: {0} - log.fine(CLASS_NAME, methodName, "500", new Object[]{this.clientId}); + // @Trace 500=Attempting to reconnect client: {0} + log.fine(CLASS_NAME, methodName, "500", new Object[] { this.clientId }); try { - connect(this.connOpts, this.userContext,new MqttReconnectActionListener(methodName)); + connect(this.connOpts, this.userContext, new MqttReconnectActionListener(methodName)); } catch (MqttSecurityException ex) { - //@TRACE 804=exception - log.fine(CLASS_NAME,methodName,"804",null, ex); + // @TRACE 804=exception + log.fine(CLASS_NAME, methodName, "804", null, ex); } catch (MqttException ex) { - //@TRACE 804=exception - log.fine(CLASS_NAME,methodName,"804",null, ex); + // @TRACE 804=exception + log.fine(CLASS_NAME, methodName, "804", null, ex); } } - - - - private void startReconnectCycle(){ + private void startReconnectCycle() { String methodName = "startReconnectCycle"; - //@Trace 503=Start reconnect timer for client: {0}, delay: {1} - log.fine(CLASS_NAME, methodName, "503", new Object[]{this.clientId, new Long(reconnectDelay)}); + // @Trace 503=Start reconnect timer for client: {0}, delay: {1} + log.fine(CLASS_NAME, methodName, "503", new Object[] { this.clientId, new Long(reconnectDelay) }); reconnectTimer = new Timer("MQTT Reconnect: " + clientId); reconnectTimer.schedule(new ReconnectTask(), reconnectDelay); } - private void stopReconnectCycle(){ + private void stopReconnectCycle() { String methodName = "stopReconnectCycle"; - //@Trace 504=Stop reconnect timer for client: {0} - log.fine(CLASS_NAME, methodName, "504", new Object[]{this.clientId}); - synchronized(clientLock) { + // @Trace 504=Stop reconnect timer for client: {0} + log.fine(CLASS_NAME, methodName, "504", new Object[] { this.clientId }); + synchronized (clientLock) { if (this.connOpts.isAutomaticReconnect()) { - if(reconnectTimer != null){ + if (reconnectTimer != null) { reconnectTimer.cancel(); reconnectTimer = null; } @@ -1203,157 +1438,173 @@ private void stopReconnectCycle(){ } } - private class ReconnectTask extends TimerTask { private static final String methodName = "ReconnectTask.run"; + public void run() { - //@Trace 506=Triggering Automatic Reconnect attempt. + // @Trace 506=Triggering Automatic Reconnect attempt. log.fine(CLASS_NAME, methodName, "506"); attemptReconnect(); } } - - class MqttReconnectCallback implements MqttCallbackExtended{ + + class MqttReconnectCallback implements MqttCallbackExtended { final boolean automaticReconnect; - MqttReconnectCallback(boolean isAutomaticReconnect){ + MqttReconnectCallback(boolean isAutomaticReconnect) { automaticReconnect = isAutomaticReconnect; } - + public void connectionLost(Throwable cause) { - if(automaticReconnect){ - // Automatic reconnect is set so make sure comms is in resting state + if (automaticReconnect) { + // Automatic reconnect is set so make sure comms is in resting + // state comms.setRestingState(true); reconnecting = true; startReconnectCycle(); } } - public void messageArrived(String topic, MqttMessage message) throws Exception {} + public void messageArrived(String topic, MqttMessage message) throws Exception { + } - public void deliveryComplete(IMqttDeliveryToken token) {} + public void deliveryComplete(IMqttDeliveryToken token) { + } + + public void connectComplete(boolean reconnect, String serverURI) { + } - public void connectComplete(boolean reconnect, String serverURI) {} - } - - class MqttReconnectActionListener implements IMqttActionListener{ - - final String methodName; - - MqttReconnectActionListener(String methodName) { - this.methodName = methodName; - } - public void onSuccess(IMqttToken asyncActionToken) { - //@Trace 501=Automatic Reconnect Successful: {0} - log.fine(CLASS_NAME, methodName, "501", new Object[]{asyncActionToken.getClient().getClientId()}); - comms.setRestingState(false); - stopReconnectCycle(); - } - - public void onFailure(IMqttToken asyncActionToken, Throwable exception) { - //@Trace 502=Automatic Reconnect failed, rescheduling: {0} - log.fine(CLASS_NAME, methodName, "502", new Object[]{asyncActionToken.getClient().getClientId()}); - if(reconnectDelay < 128000){ - reconnectDelay = reconnectDelay * 2; - } - rescheduleReconnectCycle(reconnectDelay); + class MqttReconnectActionListener implements IMqttActionListener { + + final String methodName; + + MqttReconnectActionListener(String methodName) { + this.methodName = methodName; + } + + public void onSuccess(IMqttToken asyncActionToken) { + // @Trace 501=Automatic Reconnect Successful: {0} + log.fine(CLASS_NAME, methodName, "501", new Object[] { asyncActionToken.getClient().getClientId() }); + comms.setRestingState(false); + stopReconnectCycle(); + } + + public void onFailure(IMqttToken asyncActionToken, Throwable exception) { + // @Trace 502=Automatic Reconnect failed, rescheduling: {0} + log.fine(CLASS_NAME, methodName, "502", new Object[] { asyncActionToken.getClient().getClientId() }); + if (reconnectDelay < 128000) { + reconnectDelay = reconnectDelay * 2; } - - private void rescheduleReconnectCycle(int delay){ - String reschedulemethodName = methodName+ ":rescheduleReconnectCycle"; - //@Trace 505=Rescheduling reconnect timer for client: {0}, delay: {1} - log.fine(CLASS_NAME, reschedulemethodName, "505", new Object[]{MqttAsyncClient.this.clientId, String.valueOf(reconnectDelay)}); - synchronized(clientLock) { - if(MqttAsyncClient.this.connOpts.isAutomaticReconnect()) { - if (reconnectTimer != null) { - reconnectTimer.schedule(new ReconnectTask(), delay); - } else { - // The previous reconnect timer was cancelled - reconnectDelay = delay; - startReconnectCycle(); - } + rescheduleReconnectCycle(reconnectDelay); + } + + private void rescheduleReconnectCycle(int delay) { + String reschedulemethodName = methodName + ":rescheduleReconnectCycle"; + // @Trace 505=Rescheduling reconnect timer for client: {0}, delay: + // {1} + log.fine(CLASS_NAME, reschedulemethodName, "505", + new Object[] { MqttAsyncClient.this.clientId, String.valueOf(reconnectDelay) }); + synchronized (clientLock) { + if (MqttAsyncClient.this.connOpts.isAutomaticReconnect()) { + if (reconnectTimer != null) { + reconnectTimer.schedule(new ReconnectTask(), delay); + } else { + // The previous reconnect timer was cancelled + reconnectDelay = delay; + startReconnectCycle(); } } } - + } + } - + /** * Sets the DisconnectedBufferOptions for this client - * @param bufferOpts the {@link DisconnectedBufferOptions} + * + * @param bufferOpts + * the {@link DisconnectedBufferOptions} */ public void setBufferOpts(DisconnectedBufferOptions bufferOpts) { this.comms.setDisconnectedMessageBuffer(new DisconnectedMessageBuffer(bufferOpts)); } - /** * Returns the number of messages in the Disconnected Message Buffer + * * @return Count of messages in the buffer */ - public int getBufferedMessageCount(){ + public int getBufferedMessageCount() { return this.comms.getBufferedMessageCount(); } - /** * Returns a message from the Disconnected Message Buffer - * @param bufferIndex the index of the message to be retrieved. + * + * @param bufferIndex + * the index of the message to be retrieved. * @return the message located at the bufferIndex */ - public MqttMessage getBufferedMessage(int bufferIndex){ + public MqttMessage getBufferedMessage(int bufferIndex) { return this.comms.getBufferedMessage(bufferIndex); } /** * Deletes a message from the Disconnected Message Buffer - * @param bufferIndex the index of the message to be deleted. + * + * @param bufferIndex + * the index of the message to be deleted. */ - public void deleteBufferedMessage(int bufferIndex){ + public void deleteBufferedMessage(int bufferIndex) { this.comms.deleteBufferedMessage(bufferIndex); } /** - * Returns the current number of outgoing in-flight messages - * being sent by the client. Note that this number cannot be - * guaranteed to be 100% accurate as some messages may have - * been sent or queued in the time taken for this method to return. + * Returns the current number of outgoing in-flight messages being sent by + * the client. Note that this number cannot be guaranteed to be 100% + * accurate as some messages may have been sent or queued in the time taken + * for this method to return. + * * @return the current number of in-flight messages. */ - public int getInFlightMessageCount(){ + public int getInFlightMessageCount() { return this.comms.getActualInFlight(); } - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#close() */ public void close() throws MqttException { close(false); } - - /* (non-Javadoc) + /* + * (non-Javadoc) + * * @see org.eclipse.paho.client.mqttv3.IMqttAsyncClient#close() */ public void close(boolean force) throws MqttException { final String methodName = "close"; - //@TRACE 113=< - log.fine(CLASS_NAME,methodName,"113"); + // @TRACE 113=< + log.fine(CLASS_NAME, methodName, "113"); comms.close(force); - //@TRACE 114=> - log.fine(CLASS_NAME,methodName,"114"); + // @TRACE 114=> + log.fine(CLASS_NAME, methodName, "114"); } /** * Return a debug object that can be used to help solve problems. + * * @return the {@link Debug} object */ public Debug getDebug() { - return new Debug(clientId,comms); + return new Debug(clientId, comms); } } diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/wire/MqttOutputStream.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/wire/MqttOutputStream.java index 412bf35fe..95b9cea9d 100644 --- a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/wire/MqttOutputStream.java +++ b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/internal/wire/MqttOutputStream.java @@ -87,8 +87,8 @@ public void write(MqttWireMessage message) throws IOException, MqttException { clientState.notifySentBytes(length); } - // @TRACE 500= sent {0} - log.fine(CLASS_NAME, methodName, "500", new Object[]{message}); + // @TRACE 529= sent {0} + log.fine(CLASS_NAME, methodName, "529", new Object[]{message}); } } diff --git a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/persist/MqttDefaultFilePersistence.java b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/persist/MqttDefaultFilePersistence.java index 33e03949c..e86e7e9d4 100644 --- a/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/persist/MqttDefaultFilePersistence.java +++ b/org.eclipse.paho.client.mqttv3/src/main/java/org/eclipse/paho/client/mqttv3/persist/MqttDefaultFilePersistence.java @@ -298,5 +298,6 @@ public void clear() throws MqttPersistenceException { for (int i=0; itycho-pack200b-plugin ${tycho.version} + + org.eluder.coveralls + coveralls-maven-plugin + 4.3.0 +