How to save data from MQTT subscribing to all topics

Question

I am using java processing to get sensors data from Arduino and save them into MySql database by using MQTT protocol. I succeed to publish data to my topics

• home/temperature
• home/humidity

and subscribe to all topics

• home/#

I received all the messages. Now I want to save these value into database.

My question is:

How to get the appropriate data and save it according to its topic (home/temperature & home/humidity)? thank you

this is my code:

 import org.eclipse.paho.client.mqttv3.*;
 import org.eclipse.paho.client.mqttv3.*;  
 import org.eclipse.paho.client.mqttv3.MqttCallback;
  import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
  import de.bezier.data.sql.*;
 import processing.serial.*;
  import java.util.regex.*;
  Serial myPort;
  String value;
    String broker;
   String topic;
  String topic1;
  String topic2;
  String payloadtemp;
   String payloadhum;
   MqttClient myClient;
    MySQL DB; 
    String tempsensor_ID;
     String humsensor_ID;
     MqttMessage messagetemp ; 
     MqttMessage messagehum ; 

     void setup()  
        {  
     String portlisten= Serial.list()[0];
     myPort = new Serial( this, portlisten, 115200 );
    broker   = "tcp://192.192.191.198:1883";//;
     topic = "home/#";
    topic1="home/temperatureValue" ;  
     topic2= "home/HumidityValue";
     tempsensor_ID="temp_ID";
      humsensor_ID="hum_ID";
      String user = "root";
      String pass = "";
      String database = "iot";
      DB= new MySQL(this, "localhost", database , user , pass);


          } 
       void draw(){

       if (myPort.available() > 0 ) {
        value = myPort.readString();
        String tempdata = trim(value);
         String humdata=trim(value);
            String patterntemp= "Temperature:";
           String patterntempunit="C";
            String patternhum="Humidity:";
               String patternhumunit="%";
            Pattern ptemp= Pattern.compile(Pattern.quote(patterntemp) + "             (.*?)" + Pattern.quote(patterntempunit));
         Matcher mtemp= ptemp.matcher(tempdata);
             while(mtemp.find()){
            tempdata=mtemp.group(1);
              payloadtemp=tempdata;

          }

         Pattern phum= Pattern.compile(Pattern.quote(patternhum) + "(.*?)" +               Pattern.quote(patternhumunit));
          Matcher mhum= phum.matcher(humdata);
           while(mhum.find()){
           humdata=mhum.group(1);
               payloadhum=humdata;

                }

              try {  
          myClient = new MqttClient(broker, MqttClient.generateClientId());        
     myClient.connect();        
            messagetemp = new MqttMessage((" " + payloadtemp).getBytes());  
            messagehum = new MqttMessage((" " + payloadhum).getBytes());  



             if(payloadtemp!=(null) & payloadhum!=(null)){

                 myClient.publish(topic1, messagetemp);  
                  myClient.subscribe(topic1); 

                       myClient.publish(topic2, messagehum);  
                     myClient.subscribe(topic2);          


          //myClient.subscribe(topic); //wildcard (topic) used when  subscribing to topics but not allowed when publishing a msg
                       }

                if ( DB.connect() )
               {
               myClient.setCallback(new MqttCallback() {
                  @Override
                public void messageArrived(String arg0, MqttMessage arg1) {

                     String msgrec= new String(arg1.getPayload());

                        println("RESULT " + arg0 + " - " + msgrec + " ");


                    throw new RuntimeException();

                            }

                        @Override
                     public void deliveryComplete(IMqttDeliveryToken arg0) {


                           println("DELIVERY " + arg0);

                           throw new RuntimeException();
                               }

                          @Override
                      public void connectionLost(Throwable arg0) {
                           throw new RuntimeException();
                         }
                          });


                           if(payloadtemp!=(null) & payloadhum!=(null)){


                     /*DB.execute("INSERT INTO `temperature`(`idsensor`,`temperaturevalue`) VALUES ('"+tempsensor_ID+ "','" +      messageReceived(topic1,messagetemp.getPayload()) + "');");
                    DB.execute("INSERT INTO `light`(`idsensor`,`lightvalue`) VALUES ('"+lightsensor_ID+ "','" +   messageReceived(topic2,messagehum.getPayload())+ "');");
                    */
                    }
                    }
                  else
                   {
                println("Error in the connection :-( ");
                      }  
                                 }
                        catch(MqttException mex) {  

                      System.out.println("Error: "+mex.getMessage());  
                            mex.printStackTrace();  
                     } 
                           }

                              }

Answer 1

I assume you are implementing a PAHO client....

then you need to use the parameter in the messageArrived callback to check wich topic are you getting informed about...

client.setCallback(new MqttCallback() {

      @Override
      public void messageArrived(String topicInforming, MqttMessage mqttMessage) throws Exception {
            subscriptionMessageCallback.onNewMessage(mqttMessage.toString());
      }
   ....

in this snippet topicInforming is telling you if you are getting a message abpout temperature or humidity,

for the DB you will need a connector to handle the insert transactions...

you need to define if every message should be in the same table or not, is up to you and depends about how the architecture looks like