IoT - MQTT Groovy subscriber script for updating sensor data into SQL database

The main part of the script is based on the Eclipse/Paho MQTT structure, which connects to the MQTT mosquitto broker, subscribes to the topics and payload messages sent (published) by the IoT sensors in the domotica network. The following body part shows the MQTT Client subscription handling, based on a Paho Java example by HiveMQ

Listing 1. MQTT code block



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.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import groovy.transform.Field
import groovy.sql.Sql

public abstract class Mqtt2Database implements MqttCallback {
public static void main(String[] args) {
  String mqtt_host = System.getenv("MQTT_BROKER_HOST") ;
  String mqtt_port = System.getenv("MQTT_BROKER_PORT") ;
  String tempDir = System.getenv("TEMP") ;
  int sqlok = 0 ; // result of Sql update
  println "IoT MQTT subcriber - Oracle updates: ";
  println "Environment MQTT Broker Host: $mqtt_host, Port: $mqtt_port";
  println "Temp dir : $tempDir" ;
  String mqtt_url = "tcp://$mqtt_host:$mqtt_port" ;
  println "MQTT Url: $mqtt_url" ;
  println "SQL collect MQTT settings:" 
  try {
    def sql = Sql.newInstance("jdbc:oracle:thin:@192.168.10.22:1521:xe", "<username>", "<password>", "oracle.jdbc.pool.OracleDataSource")
    sql.eachRow("select value as ipadres from micasa.settings where system='mqtt' and name='broker_ip'") {
      println "  MQTT host ${it.ipadres}"
      mqtt_host = it.ipadres
    }
    sql.eachRow("select value as port from micasa.settings where system='mqtt' and name='broker_port'") {
      println "  MQTT port: ${it.port}"
      mqtt_port = it.port
    }
  sql.close() ;
  } catch (Exception e) {
        println "SQL Exception found:" ;
        e.printStackTrace();
  }
  try {
     MqttClient client = new MqttClient(mqtt_url, "micasaSubOraSQL", new MemoryPersistence());
     client.setCallback(new MqttCallback() {
       @Override
       public void connectionLost(Throwable cause) {
       // TODO Auto-generated method stub
       println "MQTT connection lost, cause: $cause";
       }

       @Override
       public void messageArrived(String topic, MqttMessage message) throws Exception {
         String bericht = message.toString() ;   
         println "Topic: $topic - Payload Msg: $bericht " ;
         sqlok = updateDatabase(topic, bericht) ;  // sqlok = 0: no update; sqlok >=1: number of updates, usually 1
         println "Database result: $sqlok " ;
       }

       @Override
       public void deliveryComplete(IMqttDeliveryToken token) {
       // TODO Auto-generated method stub
       println "MQTT delivery complete" ;
       }
     });
     client.connect(); 
     if (!client.isConnected() ) {
          println "No MQTT broker connection" ;
     }       
     client.subscribe("micasa/+/stat/+");
  } catch (MqttException e) {
        println "MQTT Exception found:" ;
        e.printStackTrace();
  }
}
}

The interfacing with the Oracle XE database is based on the groovy.sql.SQL API, see for instance Practically Groovy: JDBC programming with Groovy for more information.

The script starts with retrieving the MQTT Broker IP-address and the Broker Port using Environment Strings like MQTT_BROKER_HOST and MQTT_BROKER_PORT. These two settings are combined into the MQTT URL 'tcp://$mqtt_host:$mqtt_port'.

You will notice that these two settings are also retrieved from the SQL database table settings. This was done in the early stages of development in order to check if the SQL connection was functioning correctly in the Groovy script.

The main class uses two methods, the first method updateDatabase supports the updating of the sensor data into the SQL database. The second method isNumericalString checks if a value is a numerical string or not, this is necessary for generating the correct update SQL statement.

Listing 2. Method updateDatabase code block



public static int updateDatabase(String msgtopic, String msgbericht) {
  int result = 0 ;
  String sensorname = "" ; // name of the sensor derived from the topic
  String sensortype = "" ; // type derived from the payload message
  String oraDateTime = ""; // date time derives from the bericht
  String sqlText = "" ;    // sql update statement
  println "  params: $msgtopic, $msgbericht"
  // parsing topic and payload message (msgbericht)
  def topicList = msgtopic.tokenize('/')
  int lng = topicList.size
  String str1 = topicList[0]
  sensorname =  topicList[1]
  String str3 = topicList[2]
  sensortype = topicList[3]
  lng = msgbericht.length()
  int lastpos = msgbericht.lastIndexOf(';')
  int firstpos = msgbericht.indexOf(';')
  println "  payload/bericht lengte: $lng, first: $firstpos, last: $lastpos"
  String part1 = ""
  if (firstpos > 0) {
    part1 =msgbericht[0..firstpos-1]
    oraDateTime = msgbericht[lastpos+1..lng-1]
    oraDateTime = oraDateTime.replace('T', ' ')
  } else {
    part1 = msgbericht ;
    def today = new Date();
    oraDateTime = String.format('%tY%>', oraDateTime)
  sqlText = sqlText.replaceAll('<>', sensorname)
  sqlText = sqlText.replaceAll('<>', sensortype)
  sqlText = sqlText.replaceAll('<>', strValue)
  sqlText = sqlText.replaceAll('<>', strValueChr)
  // println "  SQL-Text: $sqlText"
  def sql = Sql.newInstance("jdbc:oracle:thin:@192.168.10.22:1521:xe", "<username>", "<password>", "oracle.jdbc.pool.OracleDataSource")
  sql.execute(sqlText);
  result = sql.getUpdateCount()
  sql.close() ; 
  println "  SQL result: $result" 
  return result ;
}

The input parameters of the updateDatabase method are the received topic and the payload message. The topic is tokenized in order to collect the sensorname and the sensortype. The payload message contains the value of the specific sensortype like temperature or humidity. As the second part part of the message contains an optional DateTime stamp of the time that the payload message was published to the MQTT broker. Example: temp=20.5;20160131T16:30
If the DateTime stamp is omitted in the message, the current DateTime is being used.

In the second part of the method the SQL update statement is assembled into the String variable sqlText on the basis of the constant private static final String SQL_UPDATE.
The third part is to established a SQL database connection with the Oracle database, execute the assembled SQL statement, get the number of the updated rows (in our case mostly one row) and close the connection until the next received message from the MQTT broker.

In the database table present_values we store the numerical value in the database field value and also as a string in the field valuechr. If the sensor data is not numerical but contains a string value like 'on/off', 'true/false', etc., the numerical database field value is set to zero (0). Therefore in the above mentioned method we call a simple method to check as a boolean function if the received value is a numerical string. As Groovy String handling does not directly support a method to check if a string is numerical, we use in our case a simple method found in various forums on the internet and adapted it in this example by expanding with a check on a plus (+) or minus (-) sign. See listing 3.

Listing 3. Method isNumericalString code block



public static boolean isNumericString(String input) {
    boolean result = false;
    if(input != null && input.length() > 0) {
        char[] charArray = input.toCharArray();
        for(char c : charArray) {
            if ((c >= '0' && c <= '9') || (c == '.') || (c ==',') || (c == '+') || (c =='-')){
                result = true;  // it is a digit or decimal point
            } else {
                result = false;
                break;
            }
        }
    }
    return result;
}

IoT - MQTT Groovy subscriber script for updating sensor data into SQL database

Introduction

Structure of the Groovy script

Listing 1. MQTT code block

Listing 2. Method updateDatabase code block

Listing 3. Method isNumericalString code block

Complete script

Listing 4. Running the Groovy script

Output sample

Summary

Licenses and copyright

Comments or advice