mqtt service

mqtt 连接

package com.mqtttest;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;

import org.eclipse.paho.client.mqttv3.MqttCallback;

import org.eclipse.paho.client.mqttv3.MqttClient;

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.persist.MemoryPersistence;

import android.app.Service;

import android.content.Context;

import android.content.Intent;

import android.net.ConnectivityManager;

import android.net.NetworkInfo;

import android.os.Bundle;

import android.os.IBinder;

import android.util.Log;

public class MqttService extends Service {

// this is the log tag

public static final String TAG = "MqttService";

public static String ID = "";

public static String APP_KEY = "";

// the default IP address port, where your MQTT broker is running.

//private static String host = "tcp://192.168.1.51:1884";

private String host ="tcp://116.236.202.130:8090";

//private String userName = "admin";

//private String passWord = "password";

private static MqttClient client;

private static String clientID="";

private MqttConnectOptions options;

// These are the actions for the service (name are descriptive enough)

public static final String ACTION_START = "START";

public static final String ACTION_STOP = "STOP";

// Connectivity manager to determining, when the phone loses connection

private ConnectivityManager mConnMan;

@Override

public IBinder onBind(Intent arg0) {

return null;

}

// Static method to start the service

public static void actionStart(Context ctx,String key,String id) {

Log.d(TAG, "-----static method actionStart------");

Log.d(TAG, "-----static key------"+key);

Log.d(TAG, "-----static id------"+id);

if(!ID.equals(id)){

ID=id;

APP_KEY=key;

clernResounse();

}

Intent i = new Intent(ctx, MqttService.class);

i.setAction(ACTION_START);

ctx.startService(i);

}

// Static method to stop the service

public static void actionStop(Context ctx) {

Log.d(TAG, "------static method actionStop------");

ID="";

Intent i = new Intent(ctx, MqttService.class);

i.setAction(ACTION_STOP);

ctx.stopService(i);

}

@Override

public void onCreate() {

super.onCreate();

Log.d(TAG, "--------onCreate----------");

}

private void initClient() {

Log.d(TAG, "-----initClient start-----");

try {

//clientID

Log.d(TAG, "clientID:"+ID.substring(0, 20));

client = new MqttClient(host, ID.substring(0, 20), new MemoryPersistence());

options = new MqttConnectOptions();

options.setCleanSession(true);

// options.setUserName(userName);

// options.setPassword(passWord.toCharArray());

options.setConnectionTimeout(10);

options.setKeepAliveInterval(20);

client.setCallback(new MqttCallback() {

@Override

public void connectionLost(Throwable cause) {

// restart connect,when connection is lost

Log.d(TAG,"----connectionLost----------");

if (isNetworkAvailable() == true) {

connect();

} else {

Log.d(TAG, "Network is unAvailable");

sendMessageBroadcast("Network is unAvailable".toString());

}

}

@Override

public void deliveryComplete(IMqttDeliveryToken token) {

// publish ok

//System.out.println("------deliveryComplete---------" + token.isComplete());

}

@Override

public void messageArrived(String topicName, MqttMessage message)

throws Exception {

Log.d(TAG,"----messageArrived:"+message.toString());

// InputStream in=ByteToInputStream.byte2Input(message.toString().getBytes());

// FileToZip.createDirOne("d:\\test\\mqtt");

// FileToZip.saveFile(in, "d:\\test\\mqtt\\test.txt");

sendMessageBroadcast(message.toString());

}

});

Log.d(TAG, "initClient end");

} catch (Exception e) {

System.out.print(e.getMessage());

e.printStackTrace();

}

}

// Check if we are online

private boolean isNetworkAvailable() {

mConnMan = (ConnectivityManager) this.getSystemService(CONNECTIVITY_SERVICE);

NetworkInfo info = mConnMan.getActiveNetworkInfo();

if (info == null) {

return false;

}

Log.d(TAG,"check isNetworkAvailable:" + info.isConnected());

return info.isConnected();

}

@Override

public void onStart(Intent intent, int startId) {

super.onStart(intent, startId);

Log.d(TAG, "----- onStart----");

if (intent.getAction().equals(ACTION_STOP) == true) {

Log.d(TAG, "Service stop");

synchronized (this){

try {

Log.d(TAG, "Attempt to disconnect client");

if(client != null && client.isConnected()){

client.disconnect();

Log.d(TAG, "client status:" + client.isConnected());

}

client = null;

} catch (MqttException e) {

e.printStackTrace();

}

}

stopSelf();

} else if (intent.getAction().equals(ACTION_START) == true) {

Log.d(TAG, "Service start");

synchronized (this){

if(client==null){

initClient();

}

if (client.isConnected()) {

Log.d(TAG, "cur client connect stutas is true");

return;

} else {

Log.d(TAG, "cur client connect stutas is false");

Log.d(TAG, "Attempt to connection client");

connect();

}

}

}

}

//start connect

private void connect() {

Log.d(TAG, "-----start connect-----");

new Thread(new Runnable() {

@Override

public void run() {

//Log.d(TAG, "Starting connect");

try {

if(client == null){

initClient();

}

while (!client.isConnected()) {

Log.d(TAG, "connecting.........");

client.connect(options);

client.subscribe(ID, 1);

client.subscribe(APP_KEY, 1);

}

Log.d(TAG, "Starting connect status is true");

} catch (Exception e) {

e.printStackTrace();

}

}

}).start();

}

@Override

public void onDestroy() {

Log.i(TAG, "Service onDestroy");

super.onDestroy();

clernResounse();

}

private static void clernResounse(){

try {

if (client != null && client.isConnected()){

client.disconnect();

Log.d(TAG, "Service onDestroy cur client connect stutas is: "+ client.isConnected());

}

client = null;

} catch (MqttException e) {

e.printStackTrace();

}

}

/**

* sendMessageBroadcast

*/

private void sendMessageBroadcast(String message) {

Log.d(TAG,"BroadcastReceiver-------send-------" );

Bundle bundle=new Bundle();

bundle.putString("message", message);

Intent messageIntent=new Intent();

messageIntent.putExtras(bundle);

messageIntent.setAction(myActivitity.MQTT);

sendBroadcast(messageIntent);

}

}