使用Paho MqttClient在Android上嘗試使用Mqtt連接

Question

當您第一次打開應用程序時，我想要一個屏幕，您可以在其中輸入代理信息並單擊嘗試並保存。

當點擊嘗試時，它應該只顯示一個Snackbar說如果信息成功連接。

這是當按下試按鈕，我叫代碼：

private void tryConnection(View v){ 
    if(verifyInputs()){ 
     Snackbar.make(v, getString(R.string.trying_connection), Snackbar.LENGTH_LONG).show(); 

     String clientId = MqttClient.generateClientId(); 
     MqttAndroidClient client = 
       new MqttAndroidClient(this.getApplicationContext(), getServerAddress(), 
         clientId); 

     try { 
      IMqttToken token = client.connect(); 
      final View vinner = v; 
      token.setActionCallback(new IMqttActionListener() { 
       @Override 
       public void onSuccess(IMqttToken asyncActionToken) { 
        // We are connected 
        Snackbar.make(vinner, "Success", Snackbar.LENGTH_LONG).show(); 

       } 

       @Override 
       public void onFailure(IMqttToken asyncActionToken, Throwable exception) { 
        // Something went wrong e.g. connection timeout or firewall problems 
        Snackbar.make(vinner, "Fail", Snackbar.LENGTH_LONG).show(); 
       } 
      }); 
     } catch (MqttException e) { 
      e.printStackTrace(); 
     } 
    } 
} 

的問題是，onFailure似乎並不當它不能連接到服務器被調用，但是當一個連接服務器丟失。

我該如何測試連接，以便我可以存儲它並返回到主要活動？

Answer 1

好吧，我看不到您的全部服務，任何其他實現或您在哪裏/如何使用此服務，因此我提供了我的MQTT服務的示例。

也許你可以比較它，找到任何問題並修復它。

或者你也可以使用我的實現。由你決定。希望能幫助到你。

import android.app.Service; 
import android.content.Intent; 
import android.os.Binder; 
import android.os.Handler; 
import android.os.IBinder; 
import android.support.annotation.Nullable; 
import android.util.Log; 

import org.eclipse.paho.android.service.MqttAndroidClient; 
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.MqttCallback; 
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.MqttSecurityException; 
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence; 

import java.util.ArrayList; 

public class MyMqttService extends Service implements MqttCallback, IMqttActionListener { 

    private final IBinder binder = new MyBinder(); 

    private MqttAndroidClient mqttClient; 
    private MqttConnectOptions mqttConnectOptions; 
    private static final MemoryPersistence persistence = new MemoryPersistence(); 
    private ArrayList<MqttAndroidClient> lostConnectionClients; 

    private String clientId = ""; 
    private boolean isReady = false; 
    private boolean doConnectTask = true; 
    private boolean isConnectInvoked = false; 

    private Handler handler = new Handler(); 
    private final int RECONNECT_INTERVAL = 10000; // 10 seconds 
    private final int DISCONNECT_INTERVAL = 20000; // 20 seconds 
    private final int CONNECTION_TIMEOUT = 60; 
    private final int KEEP_ALIVE_INTERVAL = 200; 

    private String broker_url = "my_broker"; 

    public MyMqttService() {} 

    public class MyBinder extends Binder { 
     public MyMqttService getService() { 
      return MyMqttService.this; 
     } 
    } 

    @Nullable 
    @Override 
    public IBinder onBind(Intent intent) { 
     return binder; 
    } 

    @Override 
    public void onCreate() { 
     super.onCreate(); 
     initMqttClient(); 
    } 

    @Override 
    public void onDestroy() { 
     super.onDestroy(); 

     disconnectClients(); 
     if (isConnectInvoked && mqttClient != null && mqttClient.isConnected()) { 
      try { 
       // unsubscribe here 
       unsubscribe("¯\\_(ツ)_/¯"); 
       mqttClient.disconnect(); 
      } catch (MqttException e) { 
       Log.e("TAG", e.toString()); 
      } 
     } 

     handler.removeCallbacks(connect); 
     handler.removeCallbacks(disconnect); 
    } 

    private void initMqttClient() { 
     if(mqttClient != null) { 
      mqttClient = null; 
     } 

     lostConnectionClients = new ArrayList<>(); 

     mqttConnectOptions = new MqttConnectOptions(); 
     mqttConnectOptions.setCleanSession(true); 
     mqttConnectOptions.setConnectionTimeout(CONNECTION_TIMEOUT); 
     mqttConnectOptions.setKeepAliveInterval(KEEP_ALIVE_INTERVAL); 

     setNewMqttClient(); 

     handler.post(connect); 
     handler.postDelayed(disconnect, DISCONNECT_INTERVAL); 
    } 

    private void setNewMqttClient() { 
     mqttClient = new MqttAndroidClient(MyMqttService.this, broker_url, clientId, persistence); 
     mqttClient.setCallback(this); 
    } 

    public Runnable connect = new Runnable() { 
     public void run() { 
      connectClient(); 
      handler.postDelayed(connect, RECONNECT_INTERVAL); 
     } 
    }; 

    public Runnable disconnect = new Runnable() { 
     public void run() { 
      disconnectClients(); 
      handler.postDelayed(disconnect, DISCONNECT_INTERVAL); 
     } 
    }; 

    private void connectClient() { 
     if(doConnectTask) { 
      doConnectTask = false; 

      try { 
       isConnectInvoked = true; 
       mqttClient.connect(mqttConnectOptions, null, this); 
      } catch (MqttException ex) { 
       doConnectTask = true; 
       Log.e("TAG", ex.toString()); 
      } 
     } 
    } 

    private void disconnectClients() { 
     if (lostConnectionClients.size() > 0) { 
      // Disconnect lost connection clients 
      for (MqttAndroidClient client : lostConnectionClients) { 
       if (client.isConnected()) { 
        try { 
         client.disconnect(); 
        } catch (MqttException e) { 
         Log.e("TAG", e.toString()); 
        } 
       } 
      } 

      // Close already disconnected clients 
      for (int i = lostConnectionClients.size() - 1; i >= 0; i--) { 
       try { 
        if (!lostConnectionClients.get(i).isConnected()) { 
         MqttAndroidClient client = lostConnectionClients.get(i); 
         client.close(); 
         lostConnectionClients.remove(i); 
        } 
       } catch (IndexOutOfBoundsException e) { 
        Log.e("TAG", e.toString()); 
       } 
      } 
     } 
    } 

    @Override 
    public void deliveryComplete(IMqttDeliveryToken token) { 
     Log.e("TAG", "deliveryComplete()"); 
    } 

    @Override 
    public void messageArrived(String topic, MqttMessage message) throws Exception { 
     String payload = new String(message.getPayload()); 
     // do something 
    } 

    @Override 
    public void connectionLost(Throwable cause) { 
     Log.e("TAG", cause.getMessage()); 
    } 

    @Override 
    public void onSuccess(IMqttToken iMqttToken) { 
     isReady = true; 

     // subscribe here 
     subscribe("¯\\_(ツ)_/¯"); 
    } 

    @Override 
    public void onFailure(IMqttToken iMqttToken, Throwable throwable) { 
     setNewMqttClient(); 
     isReady = false; 
     doConnectTask = true; 
     isConnectInvoked = false; 
    } 

    private void subscribe(String topic) { 
     try { 
      mqttClient.subscribe(topic, 0); 
      isReady = true; 
     } catch (MqttSecurityException mqttSexEx) { 
      isReady = false; 
     } catch (MqttException mqttEx) { 
      isReady = false; 
     } 
    } 

    private void unsubscribe(String topic) { 
     try { 
      mqttClient.unsubscribe(topic); 
     } catch (MqttSecurityException mqttSecEx) { 
      Log.e("TAG", mqttSecEx.getMessage()); 
     } catch (MqttException mqttEx) { 
      Log.e("TAG", mqttEx.getMessage()); 
     } 
    } 

    private void publish(String topic, String jsonPayload) { 
     if(!isReady) { 
      return; 
     } 

     try { 
      MqttMessage msg = new MqttMessage(); 
      msg.setQos(0); 
      msg.setPayload(jsonPayload.getBytes("UTF-8")); 
      mqttClient.publish(topic, msg); 
     } catch (Exception ex) { 
      Log.e("TAG", ex.toString()); 
     } 
    } 
} 

我的另一個建議是隻安裝本地廣播所以，當你的活動加載和啓動服務，如果MQTT服務是能夠連接，您發送說法連接廣播和你展示一個Snackbar。如果連接失敗，則發送不同的廣播並顯示不同的消息。