android - Servizio Paho MQTT per la pubblicazione

Question

Sono nuovo per Android e servizi. Il mio obiettivo è essere in grado di impostare abbonamenti e pubblicazioni su stringhe di argomenti. Le stringhe dell'argomento e l'ID del client vengono impostati dopo l'analisi dell'inserimento dei campi di testo. Sto usando il Paho MQTT service (scaricato l'origine e costruito il JAR).

Le cause seguenti sono un'eccezione di puntatore nullo a c.publish(). Lo logcat mostra l'eccezione nel metodo IMqttDeliveryToken publish(String topic, MqttMessage message, Object userContext, IMqttActionListener callback) in MqttAndroidClient dove viene preso un token di consegna.

public class MainActivity extends Activity { 
    @Override 
    protected void onCreate(Bundle savedInstanceState) { 
     super.onCreate(savedInstanceState); 
     setContentView(R.layout.activity_main); 

     // Set locale; 
     l = getResources().getConfiguration().locale; 
    } 

    @Override 
    protected void onResume() { 
     super.onResume(); 
     addButtonListener();   
    } 

    private void addButtonListener() { 
     Button submitButton = (Button) findViewById(R.id.buttonSubmit); 

     submitButton.setOnClickListener(new OnClickListener() { 
// ... 
// validation code for fields in layout 
// ... 
// Finally, this. 

        MemoryPersistence mPer = new MemoryPersistence(); 
        String clientId = UUID.randomUUID().toString(); 
        String brokerUrl = "tcp://m2m.eclipse.org:1883"; 
        MqttAndroidClient c = new MqttAndroidClient(getApplicationContext(), brokerUrl, clientId, mPer); 
        try { 
         c.connect(); 
         String topic = "transfers/topic"; 
         String msg = "topic payload" 
         MqttMessage m = new MqttMessage(); 
         m.setPayload(msg.getBytes()); 
         m.setQos(2); 
         m.setRetained(false); 
         c.publish(topic, m); 
        } catch (MqttException e) { 
         Toast.makeText(getApplicationContext(), e.getMessage(), Toast.LENGTH_SHORT).show(); 
        } 

Potete dirmi come utilizzare il servizio per pubblicare e iscriversi? Ho fatto scorrere il progetto di esempio (da Paho Android). LWT e pubblicazione sembrano essere uniti in quanto il layout per LWT (activity_publish.xml) sembra essere utilizzato anche per la pubblicazione.

Answer 1

Il NullPointerException è perché connect() chiama un metodo asincrono ed è necessario implementare un ActionListener. In caso di successo è possibile inviare messaggi.

Log.i(LOGTAG, "MQTT Start"); 

MemoryPersistence memPer = new MemoryPersistence(); 

final MqttAndroidClient client = new MqttAndroidClient(
    context, "tcp://192.168.0.13:1883", username, memPer); 

try { 
    client.connect(null, new IMqttActionListener() { 

     @Override 
     public void onSuccess(IMqttToken mqttToken) { 
      Log.i(LOGTAG, "Client connected"); 
      Log.i(LOGTAG, "Topics="+mqttToken.getTopics()); 

      MqttMessage message = new MqttMessage("Hello, I am Android Mqtt Client.".getBytes()); 
      message.setQos(2); 
      message.setRetained(false); 

      try { 
       client.publish("messages", message); 
       Log.i(LOGTAG, "Message published"); 

       client.disconnect(); 
       Log.i(LOGTAG, "client disconnected"); 

      } catch (MqttPersistenceException e) { 
       // TODO Auto-generated catch block 
       e.printStackTrace(); 

      } catch (MqttException e) { 
       // TODO Auto-generated catch block 
       e.printStackTrace(); 
      } 
     } 

     @Override 
     public void onFailure(IMqttToken arg0, Throwable arg1) { 
      // TODO Auto-generated method stub 
      Log.i(LOGTAG, "Client connection failed: "+arg1.getMessage()); 

     } 
    }); 
} 

Answer 2

E 'molto importante capire che è necessario chiamare client.setCallback e implementare MqttCallbackHandler al fine di ricevere i messaggi sugli argomenti a cui si è iscritti.

Ecco un esempio di codice che può sia pubblicare e sottoscrivere ecc

Il codice seguente pubblica inizialmente l'argomento MQTT e payload:

• Topic: AndroidPhone
• Carico utile: Ciao, sono un client Android Mqtt.

Il codice si abbona all'argomento "tester". Se si riceve un messaggio con argomento "tester" e carico utile di "allarme attivato" allora pubblica il seguente argomento e di carico utile (attraverso il callback di cui sopra):

• Topic: Fitlet
• Carico utile: Ciao, il Broker di Mosquitto ha ricevuto il tuo messaggio dicendo che l'allarme è attivato.

Se si utilizza mosquitto quindi il seguente comando nel terminale potrebbe causare questo messaggio di essere licenziato fuori:

mosquitto_pub -h 192.168.9.100 -t tester -m "Alarm Activated" -u fred -P 1234 

Dove il mio nome utente mosquitto è fred e la password è 1234

Il Codice:

package colin.android.mqtt;  
import android.support.v7.app.AppCompatActivity; 
import android.os.Bundle; 
import android.util.Log; 
import android.widget.Toast; 
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.MqttCallbackExtended; 
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.MqttPersistenceException; 
import java.io.UnsupportedEncodingException; 

public class MainActivity extends AppCompatActivity { 
    @Override 
    protected void onCreate(Bundle savedInstanceState) { 
     super.onCreate(savedInstanceState); 
     setContentView(R.layout.activity_main); 

     String clientId = MqttClient.generateClientId(); 

     //The URL of the Mosquitto Broker is 192.168.9.100:1883 
     final MqttAndroidClient client = new MqttAndroidClient(this.getApplicationContext(), "tcp://192.168.9.100:1883", clientId); 

     client.setCallback(new MqttCallbackHandler(client));//This is here for when a message is received 

     MqttConnectOptions options = new MqttConnectOptions(); 

     try { 
      options.setUserName("fred"); 
      options.setPassword("1234".toCharArray()); 
      IMqttToken token = client.connect(options); 

      token.setActionCallback(new IMqttActionListener() { 
       @Override 
       public void onSuccess(IMqttToken asyncActionToken) { 
        // We are connected 
        Log.d("mqtt", "onSuccess"); 
//----------------------------------------------------------------------------------------------- 
        //PUBLISH THE MESSAGE      
        MqttMessage message = new MqttMessage("Hello, I am an Android Mqtt Client.".getBytes()); 
        message.setQos(2); 
        message.setRetained(false); 

        String topic = "AndroidPhone"; 

        try { 
         client.publish(topic, message); 
         Log.i("mqtt", "Message published"); 

         // client.disconnect(); 
         //Log.i("mqtt", "client disconnected"); 

        } catch (MqttPersistenceException e) { 
         // TODO Auto-generated catch block 
         e.printStackTrace(); 

        } catch (MqttException e) { 
         // TODO Auto-generated catch block 
         e.printStackTrace(); 
        } 
//----------------------------------------------------------------------------------------------- 

        String subtopic = "tester"; 
        int qos = 1; 
        try { 
         IMqttToken subToken = client.subscribe(subtopic, qos); 
         subToken.setActionCallback(new IMqttActionListener() { 
          @Override 
          public void onSuccess(IMqttToken asyncActionToken) { 
           // The message was published 
           Log.i("mqtt", "subscription success"); 
          } 

          @Override 
          public void onFailure(IMqttToken asyncActionToken, 
                Throwable exception) { 
           // The subscription could not be performed, maybe the user was not 
           // authorized to subscribe on the specified topic e.g. using wildcards 
           Log.i("mqtt", "subscription failed"); 

          } 
         }); 



        } catch (MqttException e) { 
         e.printStackTrace(); 
        } 

//--------------------------------------------------------------------------- 

       } 

       @Override 
       public void onFailure(IMqttToken asyncActionToken, Throwable exception) { 
        // Something went wrong e.g. connection timeout or firewall problems 
        Log.d("mqtt", "onFailure"); 

       } 

      }); 


     } catch (MqttException e) { 
      e.printStackTrace(); 
     } 

    } 


}//End of Activity class 

//----------------------------------------------------------------------------- 

class MqttCallbackHandler implements MqttCallbackExtended { 

    private final MqttAndroidClient client; 

    public MqttCallbackHandler (MqttAndroidClient client) 
    { 
     this.client=client; 
    } 

    @Override 
    public void connectComplete(boolean b, String s) { 
     Log.w("mqtt", s); 
    } 

    @Override 
    public void connectionLost(Throwable throwable) { 

    } 

    public void AlarmActivatedMessageReceived() 
    { 
     MqttMessage msg= new MqttMessage("Hello, the Mosquitto Broker got your message saying that the Alarm is Activated.".getBytes()); 
     try { 
      this.client.publish("Fitlet", msg); 
      Log.i("mqtt", "Message published"); 

     } catch (MqttPersistenceException e) { 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 

     } catch (MqttException e) { 
      // TODO Auto-generated catch block 
      e.printStackTrace(); 
     } 
    } 

    @Override 
    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { 
     Log.w("mqtt", mqttMessage.toString()); 

     if (mqttMessage.toString().contains("Alarm Activated")) 
     { 
      AlarmActivatedMessageReceived(); 
     } 

    } 

    @Override 
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) { 

    } 
}