MQTT——移动消息传递和 M2M 客户机开发API包

/* ============================================================================ Licensed Materials - Property of IBM 5747-SM3 (C) Copyright IBM Corp. 1999, 2012 All Rights Reserved. US Government Users Restricted Rights - Use, duplication or disclosure restricted by GSA ADP Schedule Contract with IBM Corp. ============================================================================ */ package com.ibm.msg.android.service; import java.io.File; import java.util.HashMap; import java.util.Iterator; import java.util.Map; import org.eclipse.paho.client.mqttv3.IMqttActionListener; import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.IMqttToken; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; import org.eclipse.paho.client.mqttv3.MqttCallback; import org.eclipse.paho.client.mqttv3.MqttClientPersistence; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; import org.eclipse.paho.client.mqttv3.MqttPersistenceException; import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence; import android.app.Service; import android.os.Bundle; import android.os.PowerManager; import android.os.PowerManager.WakeLock; import android.util.Log; import com.ibm.msg.android.service.MessageStore.StoredMessage; /** * <p> * This performs MQTT operations for a specific client {host,port,clientId} * </p> * <p> * Most of the major API here is intended to implement the most general forms of * the methods in IMqttAsyncClient, with slight adjustments for the Android * environment<br> * These adjustments usually consist of adding two parameters to each method :- * <ul> * <li>invocationContext - a string passed from the application to identify the * context of the operation (mainly included for support of the javascript API * implementation)</li> * <li>activityToken - a string passed from the Activity to relate back to a * callback method or other context-specific data</li> * </ul> * </p> * <p> * Operations are very much asynchronous, so success and failure are notified by * packing the relevant data into Intent objects which are broadcast back to the * Activity via the MqttService.callbackToActivity() method. * </p> */ class MqttServiceClient implements MqttCallback { // Strings for Intents etc.. private static final String TAG = "MqttServiceClient"; // Error status messages private static final String NOT_CONNECTED = "not connected"; // fields for the connection definition private String serverURI; private String clientId; private MqttClientPersistence persistence = null; private MqttConnectOptions connectOptions; // Client handle, used for callbacks... private String clientHandle; // our client object - instantiated on connect private MqttAsyncClient myClient = null; // our (parent) service object private MqttService service = null; // Saved sent messages and their corresponding Topics, activityTokens and // invocationContexts, so we can handle "deliveryComplete" callbacks // from the mqttClient private Map<IMqttDeliveryToken, String /* Topic */> savedTopics = new HashMap<IMqttDeliveryToken, String>(); private Map<IMqttDeliveryToken, MqttMessage> savedSentMessages = new HashMap<IMqttDeliveryToken, MqttMessage>(); private Map<IMqttDeliveryToken, String> savedActivityTokens = new HashMap<IMqttDeliveryToken, String>(); private Map<IMqttDeliveryToken, String> savedInvocationContexts = new HashMap<IMqttDeliveryToken, String>(); private WakeLock wakelock = null; private String wakeLockTag = null; /** * Constructor * * @param service * our "parent" service - we make callbacks to it * @param serverURI * the URI of the MQTT server to which we will connect * @param clientId * the name by which we will identify ourselves to the MQTT * server * @param persistence * @param clientHandle * the "handle" by which the activity will identify us */ MqttServiceClient(MqttService service, String serverURI, String clientId, MqttClientPersistence persistence, String clientHandle) { this.serverURI = serverURI.toString(); this.service = service; this.clientId = clientId; this.persistence = persistence; this.clientHandle = clientHandle; StringBuffer buff = new StringBuffer(this.getClass().getCanonicalName()); buff.append(" "); buff.append(clientId); buff.append(" "); buff.append("on host "); buff.append(serverURI); wakeLockTag = buff.toString(); } // The major API implementation follows :- /** * Connect to the server specified when we were instantiated * * @param options * timeout, etc * @param invocationContext * arbitrary data to be passed back to the application * @param activityToken * arbitrary identifier to be passed back to the Activity */ public void connect(MqttConnectOptions options, String invocationContext, String activityToken) { connectOptions = options; if (connectOptions.isCleanSession()) { // if it's a clean session, // discard old data service.messageStore.clearArrivedMessages(clientHandle); } service.traceDebug(TAG, "Connecting {" + serverURI + "} as {" + clientId + "}"); final Bundle resultBundle = new Bundle(); resultBundle.putString(MqttServiceConstants.CALLBACK_ACTIVITY_TOKEN, activityToken); resultBundle.putString( MqttServiceConstants.CALLBACK_INVOCATION_CONTEXT, invocationContext); resultBundle.putString(MqttServiceConstants.CALLBACK_ACTION, MqttServiceConstants.CONNECT_ACTION); try { if (persistence == null) { // ask Android where we can put files File myDir = service.getExternalFilesDir(TAG); if (myDir == null) { resultBundle.putString( MqttServiceConstants.CALLBACK_ERROR_MESSAGE, "No external storage available"); resultBundle.putSerializable( MqttServiceConstants.CALLBACK_EXCEPTION, new MqttPersistenceException()); service.callbackToActivity(clientHandle, Status.ERROR, resultBundle); return; } // use that to setup MQTT client persistence storage persistence = new MqttDefaultFilePersistence( myDir.getAbsolutePath()); } myClient = new MqttAsyncClient(serverURI, clientId, persistence); myClient.setCallback(this); IMqttActionListener listener = new MqttServiceClientListener( resultBundle) { @Override public void onSuccess(IMqttToken asyncActionToken) { //since the device's cpu can go to sleep //we may miss pings, so we get a wake lock acquireWakeLock(); service.callbackToActivity(clientHandle, Status.OK, resultBundle); deliverBacklog(); } }; myClient.connect(connectOptions, invocationContext, listener); } catch (Exception e) { handleException(resultBundle, e); } } private void handleException(final Bundle resultBundle, Exception e) { resultBundle.putString(MqttServiceConstants.CALLBACK_ERROR_MESSAGE, e.getLocalizedMessage()); resultBundle.putSerializable(MqttServiceConstants.CALLBACK_EXCEPTION, e); service.callbackToActivity(clientHandle, Status.ERROR, resultBundle); } /** * Attempt to deliver any outstanding messages we've received but which the * application hasn't acknowledged. If "cleanSession" was specified, we'll * have already