import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
private void startConnect(String clientID, String serverIP, String port) {
    //服务器地址
    String  uri ="tcp://";
    uri=uri+serverIP+":"+port;
    Log.d("MainActivity",uri+"  "+clientID);
    /**
     * 连接的选项
     */
    MqttConnectOptions conOpt = new MqttConnectOptions();
    /**设计连接超时时间*/
    conOpt.setConnectionTimeout(3000);
    /**设计心跳间隔时间300秒*/
    conOpt.setKeepAliveInterval(300);
    /**
     * 创建连接对象
     */
     client = new MqttAndroidClient(this,uri, clientID);
    /**
     * 连接后设计一个回调
     */
    client.setCallback(new MqttCallbackHandler(this, clientID));
    /**
     * 开始连接服务器，参数：ConnectionOptions,  IMqttActionListener
     */
    try {
        client.connect(conOpt, null, new ConnectCallBackHandler(this));
    } catch (MqttException e) {
        e.printStackTrace();
    }

}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public void testMsg() throws MqttException {
    final MqttAndroidClient client = RxMqtt.client(RuntimeEnvironment.application, url);
    TestObserver observer = RxMqtt.message(client, topic).doOnNext(new Consumer<MqttMessage>() {
        @Override
        public void accept(MqttMessage msg) throws Exception {
            System.out.println(msg);
        }
    }).test();
    try {
        observer.await(10, SECONDS);
    } catch (InterruptedException e) {
        fail(e.toString());
        e.printStackTrace();
    }
    observer.assertNoErrors();
    //assertTrue(true);
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public MqttAndroidClient createClient(String id, String serverURI, String clientId) {
    MqttClientPersistence mqttClientPersistence = new MemoryPersistence();
    MqttAndroidClient client = new MqttAndroidClient(MyApplication.getContext(), serverURI, clientId, mqttClientPersistence);
    client.setCallback(new MqttCallback() {
        @Override
        public void connectionLost(Throwable cause) {
            LogUtil.e("connectionLost");
            EventBus.getDefault().post(new MQTTActionEvent(Constant.MQTTStatusConstant.CONNECTION_LOST, null, cause));

        }

        @Override
        public void messageArrived(String topic, MqttMessage message) throws Exception {
            LogUtil.d("topic is " + topic + ",message is " + message.toString() + ", qos is " + message.getQos());
            EventBus.getDefault().postSticky(new MessageEvent(new EmqMessage(topic, message)));

        }

        @Override
        public void deliveryComplete(IMqttDeliveryToken token) {
            LogUtil.d("deliveryComplete");


        }
    });

    mClients.put(id, client);

    return client;

}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public boolean disconnect(MqttAndroidClient client) {
    if (!isConnected(client)) {
        return true;
    }

    try {
        client.disconnect();
        return true;
    } catch (MqttException e) {
        e.printStackTrace();
        return false;
    }
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public MQTTconect(Context contexto,MqttCallback mqttcallback) {

        //gera um código randômico que serve como identificação do cliente
        clientID = MqttClient.generateClientId()+"circularUFPAapp";
        //cria um objeto MQTTClient android entregando como parametro o endereço o servidor e o id do cliente
        mqttAndroidClient = new MqttAndroidClient(contexto, serverAndress, clientID);
        //configura um objeto CallBack (objeto de chamada caso haja alteração)
        mqttAndroidClient.setCallback(mqttcallback);


    }
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
/**
 * Creates a connection from persisted information in the database store, attempting
 * to create a {@link MqttAndroidClient} and the client handle.
 * @param clientId The id of the client
 * @param host the server which the client is connecting to
 * @param port the port on the server which the client will attempt to connect to
 * @param context the application context
 * @param tlsConnection true if the connection is secured by SSL
 * @return a new instance of <code>Connection</code>
 */
public static Connection createConnection(String clientHandle, String clientId, String host, int port, Context context, boolean tlsConnection){

    String uri;
    if(tlsConnection) {
        uri = "ssl://" + host + ":" + port;
    } else {
        uri = "tcp://" + host + ":" + port;
    }

    executor = Executors.newFixedThreadPool(1);

    MqttAndroidClient client = new MqttAndroidClient(context, uri, clientId);
    return new Connection(clientHandle, clientId, host, port, context, client, tlsConnection);
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public void updateConnection(String clientId, String host, int port, boolean tlsConnection){
    String uri;
    if(tlsConnection) {
        uri = "ssl://" + host + ":" + port;
    } else {
        uri = "tcp://" + host + ":" + port;
    }

    this.clientId = clientId;
    this.host = host;
    this.port = port;
    this.tlsConnection = tlsConnection;
    this.client = new MqttAndroidClient(context, uri, clientId);
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
/**
 * Creates a connection object with the server information and the client
 * hand which is the reference used to pass the client around activities
 * @param clientHandle The handle to this <code>Connection</code> object
 * @param clientId The Id of the client
 * @param host The server which the client is connecting to
 * @param port The port on the server which the client will attempt to connect to
 * @param context The application context
 * @param client The MqttAndroidClient which communicates with the service for this connection
 * @param tlsConnection true if the connection is secured by SSL
 */
private Connection(String clientHandle, String clientId, String host,
                   int port, Context context, MqttAndroidClient client, boolean tlsConnection) {
    //generate the client handle from its hash code
    this.clientHandle = clientHandle;
    this.clientId = clientId;
    this.host = host;
    this.port = port;
    this.context = context;
    this.client = client;
    this.tlsConnection = tlsConnection;
    history = new ArrayList<String>();
    String sb = "Client: " +
            clientId +
            " created";
    addAction(sb);

    try {
        sparkplugMetrics.put("Analog 1", new MetricBuilder("Analog 1", MetricDataType.Double, 0D).createMetric());
        sparkplugMetrics.put("Analog 2", new MetricBuilder("Analog 2", MetricDataType.Double, 0D).createMetric());
        sparkplugMetrics.put("Analog 3", new MetricBuilder("Analog 3", MetricDataType.Double, 0D).createMetric());
        sparkplugMetrics.put("Analog 4", new MetricBuilder("Analog 4", MetricDataType.Double, 0D).createMetric());
        sparkplugMetrics.put("Boolean 1", new MetricBuilder("Boolean 1", MetricDataType.Boolean, false).createMetric());
        sparkplugMetrics.put("Boolean 2", new MetricBuilder("Boolean 2", MetricDataType.Boolean, false).createMetric());
        sparkplugMetrics.put("Boolean 3", new MetricBuilder("Boolean 3", MetricDataType.Boolean, false).createMetric());
        sparkplugMetrics.put("Boolean 4", new MetricBuilder("Boolean 4", MetricDataType.Boolean, false).createMetric());

        sparkplugMetrics.put("Scan Code", new MetricBuilder("Scan Code", MetricDataType.String, "").createMetric());
    } catch (Exception e) {
        Log.e(TAG, "Failed to set up metrics", e);
    }

    getGpsLocation();
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
/**
 * Auto close client
 * mqttConnectOptions.userName = it }$
 * mqttConnectOptions.password = it.toCharArray() }
 * @param client
 * @param topic
 * @return
 */
@NonNull
@CheckReturnValue
public static Observable<MqttMessage> remessage(@NonNull final MqttAndroidClient client,
                                                @NonNull final String topic) {
    final Observable<MqttMessage> msgObs =
            Observable.create(new ObservableOnSubscribe<MqttMessage>() {
        public void subscribe(
                @NonNull final ObservableEmitter<MqttMessage> emitter) throws Exception {
            client.subscribe(topic, 0, new IMqttMessageListener() {
                @Override
                public void messageArrived(
                        String topic2, @NonNull final MqttMessage message) throws Exception {
                    if (!emitter.isDisposed()) {
                        emitter.onNext(message);
                    }
                }
            });
        }
    });

    if (client.isConnected()) {
        return msgObs;
    } else {
        return reconnect(client).flatMapObservable(
                new Function<IMqttToken, ObservableSource<MqttMessage>>() {
            @Override
            public ObservableSource<MqttMessage> apply(IMqttToken token) throws Exception {
                return msgObs;
            }
        });
    }
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
@NonNull
@CheckReturnValue
public static Maybe<IMqttToken> connect(
        @NonNull final MqttAndroidClient client,
        @NonNull final DisconnectedBufferOptions disconnectedBufferOptions) {
    return connect(client, new MqttConnectOptions(), disconnectedBufferOptions);
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
@Test
public void testConnect() throws MqttException {
    final MqttAndroidClient client = RxMqtt.client(RuntimeEnvironment.application, url);
    TestObserver observer = RxMqtt.connect(client).doOnNext(new Consumer<IMqttToken>() {
        @Override
        public void accept(IMqttToken token) throws Exception {
            System.out.println(token);
        }
    }).test();
    observer.awaitTerminalEvent();
    observer.assertTerminated();
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public StatusListener(Context ctx, MqttAndroidClient client) {
    this.ctx = ctx;
    this.client = client;
    CharSequence text = "Connection Failure";  //default
    int duration = Toast.LENGTH_SHORT;
    toast = Toast.makeText(ctx, text, duration);

}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
private void disconnectAllClient() {
    for (MqttAndroidClient client : mClients.values()) {
        disconnect(client);
    }
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public boolean isConnected(MqttAndroidClient client) {
    return client != null && client.isConnected();
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public MqttAndroidClient getClient(String id) {
    return mClients.get(id);
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public void setMQTTclient(MqttAndroidClient newMqttClient) {
    this.myMQTTclient = newMqttClient;
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public MqttAndroidClient getMyMQTTclient() {
    return this.myMQTTclient;
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
public void createMqttClient(String serverAddress)
{
    String mqttClientId = MqttClient.generateClientId();
    System.out.println("Server address: " + serverAddress);
    myMQTTclient = new MqttAndroidClient(this.getApplicationContext(), serverAddress, mqttClientId);
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
/**
 * 获取MqttAndroidClient实例
 * @return
 */
public static MqttAndroidClient getMqttAndroidClientInstace(){
    if(client!=null)
        return  client;
    return null;
}
 

import org.eclipse.paho.android.service.MqttAndroidClient; //导入依赖的package包/类
@Override
protected void onCreate(Bundle savedInstanceState) {
    super.onCreate(savedInstanceState);
    setContentView(R.layout.activity_publish);
    ButterKnife.bind(this);
    setTitle("模拟服务器发布主题");
    initDate();

    /**1.开始发布*/
    btnStartPub.setOnClickListener(new View.OnClickListener() {
        @Override
        public void onClick(View view) {
            /**获取发布的主题*/
            pubTopic = edPubTopic.getText().toString().trim();
            /**获取发布的消息*/
            pubMessage = edPubMessage.getText().toString().trim();
            /**消息的服务质量*/
            int qos=0;
            /**消息是否保持*/
            boolean retain=false;
            /**要发布的消息内容*/
            byte[] message=pubMessage.getBytes();
            if(pubTopic!=null&&!"".equals(pubTopic)){
                /**获取client对象*/
                MqttAndroidClient client = MainActivity.getMqttAndroidClientInstace();
                if(client!=null){
                    try {
                        /**发布一个主题:如果主题名一样不会新建一个主题，会复用*/
                        client.publish(pubTopic,message,qos,retain,null,new PublishCallBackHandler(PublishActivity.this));
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                }else{
                    Log.e(PA,"MqttAndroidClient==null");
                }
            }else{
                Toast.makeText(PublishActivity.this,"发布的主题不能为空",Toast.LENGTH_SHORT).show();
            }
        }
    });
}