Merge pull request #6 from hannesa2/Erguotou/develop

Erguotou/develop

Author: hannesa2

build.gradle

@@ -21,7 +21,7 @@ ext {
     serviceArchivesBaseName = 'org.eclipse.paho.android.service'
     serviceVersion = '1.2.0'
 
-    clientVersion = '1.2.0'
+    clientVersion = '1.2.5'
 
     mavenUrl = "https://repo.eclipse.org/content/repositories/paho-releases/"
 }

org.eclipse.paho.android.service/src/main/java/org/eclipse/paho/android/service/MqttAndroidClient.java

@@ -1028,6 +1028,33 @@ public IMqttToken unsubscribe(String[] topic, Object userContext, IMqttActionLis
         return token;
     }
 
+    /**
+     * Removes a published message corresponding to the token.
+     * <p>If a publish is requested with QoS1 or Qos2 and the publish callback is
+     * not called yet, this function returns true, the publish called will never
+     * be called, and a messageId corresponding to the token will become reusable.
+     * </p>
+     * <p>If the publish callback is already be called, this function returns false.
+     * </p>
+     * <p>This function might not stop sending the published message.
+     * </p>
+     * *
+     *
+     * @param token the token of removing published message
+     * @return if the message is removed then true, otherwise false
+     * @throws MqttException if there was an error removing the message.
+     */
+    @Override
+    public boolean removeMessage(IMqttDeliveryToken token) throws MqttException {
+        if (token.getMessage().getQos() > 0 && !token.isComplete())
+            return false;
+        int i = tokenMap.indexOfValue(token);
+        if (i > -1) tokenMap.removeAt(i);
+        if (callback != null)
+            callback.deliveryComplete(token);
+        return true;
+    }
+
     /**
      * Returns the delivery tokens for any outstanding publish operations.
      * <p>
@@ -1176,6 +1203,16 @@ public void setManualAcks(boolean manualAcks) {
         throw new UnsupportedOperationException();
     }
 
+    /**
+     * Will attempt to reconnect to the server after the client has lost connection.
+     *
+     * @throws MqttException if an error occurs attempting to reconnect
+     */
+    @Override
+    public void reconnect() throws MqttException {
+        mqttService.reconnect();
+    }
+
     /**
      * Process the results of a connection
      *
@@ -1405,6 +1442,19 @@ public void deleteBufferedMessage(int bufferIndex) {
         mqttService.deleteBufferedMessage(clientHandle, 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.
+     *
+     * @return the current number of in-flight messages.
+     */
+    @Override
+    public int getInFlightMessageCount() {
+        return mqttService.getInFlightMessageCount(clientHandle);
+    }
+
     /**
      * Get the SSLSocketFactory using SSL key store and password
      * <p>

org.eclipse.paho.android.service/src/main/java/org/eclipse/paho/android/service/MqttConnection.java

@@ -710,21 +710,10 @@ public void deliveryComplete(IMqttDeliveryToken messageToken) {
 
         service.traceDebug(TAG, "deliveryComplete(" + messageToken + ")");
 
-        MqttMessage message = savedSentMessages.remove(messageToken);
-        if (message != null) { // If I don't know about the message, it's
-            // irrelevant
-            String topic = savedTopics.remove(messageToken);
-            String activityToken = savedActivityTokens.remove(messageToken);
-            String invocationContext = savedInvocationContexts.remove(messageToken);
-
-            Bundle resultBundle = messageToBundle(null, topic, message);
-            if (activityToken != null) {
-                resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION, MqttServiceConstants.SEND_ACTION);
-                resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN, activityToken);
-                resultBundle.putString(MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, invocationContext);
-
+        Bundle resultBundle = popSendDetails(messageToken);
+        if (resultBundle != null) {
+            if (MqttServiceConstants.SEND_ACTION.equals(resultBundle.getString(MqttServiceConstants.CALLBACK_ACTION)))
                 service.callbackToActivity(clientHandle, Status.OK, resultBundle);
-            }
             resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION, MqttServiceConstants.MESSAGE_DELIVERED_ACTION);
             service.callbackToActivity(clientHandle, Status.OK, resultBundle);
         }
@@ -751,6 +740,29 @@ public void messageArrived(String topic, MqttMessage message) throws Exception {
         service.callbackToActivity(clientHandle, Status.OK, resultBundle);
     }
 
+    /**
+     * Removed store details of sent messages in "deliveryComplete"
+     * callbacks from the mqttClient
+     */
+    private synchronized Bundle popSendDetails(final IMqttDeliveryToken messageToken) {
+        MqttMessage message = savedSentMessages.remove(messageToken);
+        if (message != null) { // If I don't know about the message, it's
+            // irrelevant
+            String topic = savedTopics.remove(messageToken);
+            String activityToken = savedActivityTokens.remove(messageToken);
+            String invocationContext = savedInvocationContexts.remove(messageToken);
+
+            Bundle resultBundle = messageToBundle(null, topic, message);
+            if (activityToken != null) {
+                resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION, MqttServiceConstants.SEND_ACTION);
+                resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN, activityToken);
+                resultBundle.putString(MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, invocationContext);
+            }
+            return resultBundle;
+        }
+        return null;
+    }
+
     /**
      * Store details of sent messages so we can handle "deliveryComplete"
      * callbacks from the mqttClient
@@ -761,7 +773,7 @@ public void messageArrived(String topic, MqttMessage message) throws Exception {
      * @param invocationContext
      * @param activityToken
      */
-    private void storeSendDetails(final String topic, final MqttMessage msg, final IMqttDeliveryToken messageToken,
+    private synchronized void storeSendDetails(final String topic, final MqttMessage msg, final IMqttDeliveryToken messageToken,
             final String invocationContext, final String activityToken) {
         savedTopics.put(messageToken, topic);
         savedSentMessages.put(messageToken, msg);
@@ -920,6 +932,10 @@ public void deleteBufferedMessage(int bufferIndex) {
         myClient.deleteBufferedMessage(bufferIndex);
     }
 
+    public int getInFlightMessageCount() {
+        return myClient.getInFlightMessageCount();
+    }
+
     /**
      * General-purpose IMqttActionListener for the Client context
      * <p>

org.eclipse.paho.android.service/src/main/java/org/eclipse/paho/android/service/MqttService.java

@@ -762,6 +762,11 @@ public void deleteBufferedMessage(String clientHandle, int bufferIndex) {
         client.deleteBufferedMessage(bufferIndex);
     }
 
+    public int getInFlightMessageCount(String clientHandle) {
+        MqttConnection client = getConnection(clientHandle);
+        return client.getInFlightMessageCount();
+    }
+
     /*
      * Called in response to a change in network connection - after losing a
      * connection to the server, this allows us to wait until we have a usable