基本信息
源码名称:mqtt 实时消息推送实例源码下载(android)
源码大小:1.31M
文件格式:.rar
开发语言:Java
更新时间:2017-05-27
友情提示:(无需注册或充值,赞助后即可获取资源下载链接)
嘿,亲!知识可是无价之宝呢,但咱这精心整理的资料也耗费了不少心血呀。小小地破费一下,绝对物超所值哦!如有下载和支付问题,请联系我们QQ(微信同号):813200300
本次赞助数额为: 2 元×
微信扫码支付:2 元
×
请留下您的邮箱,我们将在2小时内将文件发到您的邮箱
源码介绍
package com.tokudu.demo; import java.io.IOException; import android.app.AlarmManager; import android.app.NotificationManager; import android.app.PendingIntent; import android.content.BroadcastReceiver; import android.content.Context; import android.content.Intent; import android.content.IntentFilter; import android.content.SharedPreferences; import android.net.ConnectivityManager; import android.net.NetworkInfo; import android.os.AsyncTask; import android.util.Log; import com.ibm.mqtt.IMqttClient; import com.ibm.mqtt.MqttClient; import com.ibm.mqtt.MqttException; import com.ibm.mqtt.MqttPersistence; import com.ibm.mqtt.MqttPersistenceException; import com.ibm.mqtt.MqttSimpleCallback; public class ConnectServerTask extends AsyncTask<Object, Void, Object> { // this is the log tag public static final String TAG = "DemoPushService"; // This is the instance of an MQTT connection. public MQTTConnection mConnection; // the IP address, where your MQTT broker is running. // private static final String MQTT_HOST = "192.168.1.71"; private static final String MQTT_HOST = "192.168.1.18"; // private static final String MQTT_HOST = "192.168.1.190"; // the port at which the broker is running. private static int MQTT_BROKER_PORT_NUM = 1883; // Let's not use the MQTT persistence. private static MqttPersistence MQTT_PERSISTENCE = null; // We don't need to remember any state between the connections, so we use a // clean start. private static boolean MQTT_CLEAN_START = true; // Let's set the internal keep alive for MQTT to 15 mins. I haven't tested // this value much. It could probably be increased. private static short MQTT_KEEP_ALIVE = 60 * 15; // Set quality of services to 0 (at most once delivery), since we don't want // push notifications // arrive more than once. However, this means that some messages might get // lost (delivery is not guaranteed) private static int[] MQTT_QUALITIES_OF_SERVICE = { 0 }; private static int MQTT_QUALITY_OF_SERVICE = 0; // The broker should not retain any messages. private static boolean MQTT_RETAINED_PUBLISH = false; // MQTT client ID, which is given the broker. In this example, I also use // this for the topic header. // You can use this to run push notifications for multiple apps with one // MQTT broker. public static String MQTT_CLIENT_ID = "tokudu"; // These are the actions for the service (name are descriptive enough) private static final String ACTION_START = MQTT_CLIENT_ID ".START"; private static final String ACTION_STOP = MQTT_CLIENT_ID ".STOP"; private static final String ACTION_KEEPALIVE = MQTT_CLIENT_ID ".KEEP_ALIVE"; private static final String ACTION_RECONNECT = MQTT_CLIENT_ID ".RECONNECT"; // Connection log for the push service. Good for debugging. private ConnectionLog mLog; // Connectivity manager to determining, when the phone loses connection private ConnectivityManager mConnMan; // Notification manager to displaying arrived push notifications private NotificationManager mNotifMan; // Whether or not the service has been started. private boolean mStarted; // This the application level keep-alive interval, that is used by the // AlarmManager // to keep the connection active, even when the device goes to sleep. private static final long KEEP_ALIVE_INTERVAL = 1000 * 60 * 28; // Retry intervals, when the connection is lost. private static final long INITIAL_RETRY_INTERVAL = 1000 * 10; private static final long MAXIMUM_RETRY_INTERVAL = 1000 * 60 * 30; // Preferences instance private SharedPreferences mPrefs; // We store in the preferences, whether or not the service has been started public static final String PREF_STARTED = "isStarted"; // We also store the deviceID (target) public static final String PREF_DEVICE_ID = "deviceID"; // We store the last retry interval public static final String PREF_RETRY = "retryInterval"; // Notification title public static String NOTIF_TITLE = "Tokudu"; // Notification id private static int NOTIF_CONNECTED = 0; private long mStartTime; public interface ConnectionCallBack { /** 处理接收到的信息 */ public void handleMsg(String msg); } private ConnectionCallBack callBack; private Context context; public ConnectServerTask(Context context, ConnectionCallBack callBack) { mStartTime = System.currentTimeMillis(); this.context = context; this.callBack = callBack; mPrefs = context.getSharedPreferences(TAG, Context.MODE_PRIVATE); } public void setCallBack(ConnectionCallBack callBack) { this.callBack = callBack; } @Override protected Object doInBackground(Object... params) { start(); return null; } @Override protected void onPreExecute() { super.onPreExecute(); } @Override protected void onPostExecute(Object result) { super.onPostExecute(result); } private void log(String message, Throwable e) { if (e != null) { Log.e(TAG, message, e); } else { Log.i(TAG, message); } if (mLog != null) { try { mLog.println(message); } catch (IOException ex) { } } } // Sets whether or not the services has been started in the preferences. private void setStarted(boolean started) { mPrefs.edit().putBoolean(PREF_STARTED, started).commit(); mStarted = started; } // We schedule a reconnect based on the starttime of the service public void scheduleReconnect(long startTime) { // the last keep-alive interval long interval = mPrefs.getLong(PREF_RETRY, INITIAL_RETRY_INTERVAL); // Calculate the elapsed time since the start long now = System.currentTimeMillis(); long elapsed = now - startTime; // Set an appropriate interval based on the elapsed time since start if (elapsed < interval) { interval = Math.min(interval * 4, MAXIMUM_RETRY_INTERVAL); } else { interval = INITIAL_RETRY_INTERVAL; } log("Rescheduling connection in " interval "ms."); // Save the new internval mPrefs.edit().putLong(PREF_RETRY, interval).commit(); // Schedule a reconnect using the alarm manager. Intent i = new Intent(); i.setClass(context, PushService2.class); i.setAction(ACTION_RECONNECT); PendingIntent pi = PendingIntent.getService(context, 0, i, 0); AlarmManager alarmMgr = (AlarmManager) context .getSystemService(Context.ALARM_SERVICE); alarmMgr.set(AlarmManager.RTC_WAKEUP, now interval, pi); } // log helper function private void log(String message) { log(message, null); } // Check if we are online private boolean isNetworkAvailable() { NetworkInfo info = mConnMan.getActiveNetworkInfo(); if (info == null) { return false; } return info.isConnected(); } // This inner class is a wrapper on top of MQTT client. class MQTTConnection implements MqttSimpleCallback { IMqttClient mqttClient = null; ConnectionCallBack callBack = null; // Creates a new connection given the broker address and initial topic public MQTTConnection(String brokerHostName, String initTopic,ConnectionCallBack callBack) throws MqttException { this.callBack = callBack; // Create connection spec String mqttConnSpec = "tcp://" brokerHostName "@" MQTT_BROKER_PORT_NUM; // Create the client and connect mqttClient = MqttClient.createMqttClient(mqttConnSpec, MQTT_PERSISTENCE); String clientID = MQTT_CLIENT_ID "/" mPrefs.getString(PREF_DEVICE_ID, ""); mqttClient.connect(clientID, MQTT_CLEAN_START, MQTT_KEEP_ALIVE); // register this client app has being able to receive messages mqttClient.registerSimpleHandler(this); // Subscribe to an initial topic, which is combination of client ID // and device ID. initTopic = MQTT_CLIENT_ID "/" initTopic; subscribeToTopic(initTopic); log("Connection established to " brokerHostName " on topic " initTopic); // Save start time mStartTime = System.currentTimeMillis(); // Star the keep-alives startKeepAlives(); } // Disconnect public void disconnect() { try { stopKeepAlives(); mqttClient.disconnect(); } catch (MqttPersistenceException e) { log("MqttException" (e.getMessage() != null ? e.getMessage() : " NULL"), e); } } /* * Send a request to the message broker to be sent messages published * with the specified topic name. Wildcards are allowed. */ private void subscribeToTopic(String topicName) throws MqttException { if ((mqttClient == null) || (mqttClient.isConnected() == false)) { // quick sanity check - don't try and subscribe if we don't have // a connection log("Connection error" "No connection"); } else { String[] topics = { topicName }; mqttClient.subscribe(topics, MQTT_QUALITIES_OF_SERVICE); } } /* * Sends a message to the message broker, requesting that it be * published to the specified topic. */ private void publishToTopic(String topicName, String message) throws MqttException { if ((mqttClient == null) || (mqttClient.isConnected() == false)) { // quick sanity check - don't try and publish if we don't have // a connection log("No connection to public to"); } else { mqttClient.publish(topicName, message.getBytes(), MQTT_QUALITY_OF_SERVICE, MQTT_RETAINED_PUBLISH); } } /* * Called if the application loses it's connection to the message * broker. */ public void connectionLost() throws Exception { log("Loss of connection" "connection downed"); stopKeepAlives(); // null itself mConnection = null; if (isNetworkAvailable() == true) { reconnectIfNecessary(); } } // Remove all scheduled keep alives void stopKeepAlives() { Intent i = new Intent(); i.setClass(context, PushService2.class); i.setAction(ACTION_KEEPALIVE); PendingIntent pi = PendingIntent.getService(context, 0, i, 0); AlarmManager alarmMgr = (AlarmManager) context .getSystemService(Context.ALARM_SERVICE); alarmMgr.cancel(pi); } /* * Called when we receive a message from the message broker. */ public void publishArrived(String topicName, byte[] payload, int qos, boolean retained) { // Show a notification String s = new String(payload); //showNotification(s); if(callBack!=null){ callBack.handleMsg(s); } log("Got message: " s); } public void sendKeepAlive() throws MqttException { log("Sending keep alive"); // publish to a keep-alive topic publishToTopic(MQTT_CLIENT_ID "/keepalive", mPrefs.getString(PREF_DEVICE_ID, "")); } } // Schedule application level keep-alives using the AlarmManager private void startKeepAlives() { Intent i = new Intent(); i.setClass(context, PushService2.class); i.setAction(ACTION_KEEPALIVE); PendingIntent pi = PendingIntent.getService(context, 0, i, 0); AlarmManager alarmMgr = (AlarmManager) context .getSystemService(Context.ALARM_SERVICE); alarmMgr.setRepeating(AlarmManager.RTC_WAKEUP, System.currentTimeMillis() KEEP_ALIVE_INTERVAL, KEEP_ALIVE_INTERVAL, pi); } public synchronized void reconnectIfNecessary() { if (mStarted == true && mConnection == null) { log("Reconnecting..."); connect(); } } // This receiver listeners for network changes and updates the MQTT // connection // accordingly private BroadcastReceiver mConnectivityChanged = new BroadcastReceiver() { @Override public void onReceive(Context context, Intent intent) { // Get network info NetworkInfo info = (NetworkInfo) intent .getParcelableExtra(ConnectivityManager.EXTRA_NETWORK_INFO); // Is there connectivity? boolean hasConnectivity = (info != null && info.isConnected()) ? true : false; log("Connectivity changed: connected=" hasConnectivity); if (hasConnectivity) { reconnectIfNecessary(); } else if (mConnection != null) { // if there no connectivity, make sure MQTT connection is // destroyed mConnection.disconnect(); cancelReconnect(); mConnection = null; } } }; // Remove the scheduled reconnect public void cancelReconnect() { Intent i = new Intent(); i.setClass(context, PushService2.class); i.setAction(ACTION_RECONNECT); PendingIntent pi = PendingIntent.getService(context, 0, i, 0); AlarmManager alarmMgr = (AlarmManager) context .getSystemService(Context.ALARM_SERVICE); alarmMgr.cancel(pi); } public synchronized void start() { log("Starting service..."); // Do nothing, if the service is already running. if (mStarted == true) { Log.w(TAG, "Attempt to start connection that is already active"); return; } // Establish an MQTT connection connect(); // Register a connectivity listener context.registerReceiver(mConnectivityChanged, new IntentFilter( ConnectivityManager.CONNECTIVITY_ACTION)); } public synchronized void stop() { // Do nothing, if the service is not running. if (mStarted == false) { Log.w(TAG, "Attempt to stop connection not active."); return; } // Save stopped state in the preferences setStarted(false); // Remove the connectivity receiver context.unregisterReceiver(mConnectivityChanged); // Any existing reconnect timers should be removed, since we explicitly // stopping the service. cancelReconnect(); // Destroy the MQTT connection if there is one if (mConnection != null) { mConnection.disconnect(); mConnection = null; } } public synchronized void keepAlive() { try { // Send a keep alive, if there is a connection. if (mStarted == true && mConnection != null) { mConnection.sendKeepAlive(); } } catch (MqttException e) { log("MqttException: " (e.getMessage() != null ? e.getMessage() : "NULL"), e); mConnection.disconnect(); mConnection = null; cancelReconnect(); } } // private synchronized void connect() { log("Connecting..."); // fetch the device ID from the preferences. String deviceID = mPrefs.getString(PREF_DEVICE_ID, null); // Create a new connection only if the device id is not NULL if (deviceID == null) { log("Device ID not found."); } else { try { mConnection = new MQTTConnection(MQTT_HOST, deviceID,callBack); } catch (MqttException e) { e.printStackTrace(); // Schedule a reconnect, if we failed to connect log("MqttException: " (e.getMessage() != null ? e.getMessage() : "NULL")); if (isNetworkAvailable()) { scheduleReconnect(mStartTime); } } setStarted(true); } } }