title: Storing MQTT data into a Mongo database
body: <p>I wrote a program which connects to a MQTT Broker and stores every MQTT message to a Mongo database. It works fine but I'm really not sure I've done it the right way. It opens lots of connections to the mongo server.</p>
  
  <p>Here is the code :</p>
  
  <pre><code>public class MqttConnection {
  
  private static final String BROKER_IP = Preferences.getProperty("mqttbroker-ip");
  private static final String BROKER_PORT = Preferences.getProperty("mqttbroker-port");
  private static final String BROKER_URI = "tcp://" + BROKER_IP + ":" + BROKER_PORT;
  
  private static final String CLIENT_ID = Preferences.getProperty("mqttbroker-clientid");
  
  private static final String USERNAME = Preferences.getProperty("mqttbroker-username");
  private static final String PASSWORD = Preferences.getProperty("mqttbroker-password");
  
  private static final String TOPIC = Preferences.getProperty("mqttbroker-topic");
  
  private static final SimpleDateFormat SDF = new SimpleDateFormat("YYYYMMddHH:mm:ss");
  
  private static MqttClient client;
  private static final MqttConnectOptions options;
  
  private static final MongoClient mongoClient;
  
  public enum Type {
      Double, Boolean, String;
  }
  static {
      mongoClient = MongoConnection.getInstance();
      try {
          client = new MqttClient(BROKER_URI, CLIENT_ID);
      } catch (MqttException ex) {
          Logger.getLogger(MqttConnection.class.getName()).log(Level.SEVERE, null, ex);
      }
      options = new MqttConnectOptions();
      options.setUserName(USERNAME);
      options.setPassword(PASSWORD.toCharArray());
  }
  
  public static void connect() {
      try {
          if (Preferences.getProperty("mqttbroker-isAuth").equalsIgnoreCase("true")) {
              client.connect(options);
          } else {
              client.connect();
          }
          System.out.println(SDF.format(new Date()) + " - MQTT - Connecté au broker");
          client.setCallback(new ConnectionCallback());
          client.subscribe(TOPIC);
      } catch (MqttException ex) {
          System.out.println(SDF.format(new Date()) + " - MQTT - Erreur de connexion... Nouvelle tentative dans 5 secondes...");
          if (!client.isConnected()) {
              try {
                  Thread.sleep(5000);
              } catch (InterruptedException ex1) {
                  Logger.getLogger(MqttConnection.class.getName()).log(Level.SEVERE, null, ex1);
              }
              connect();
          }
      }
  }
  
  static class ConnectionCallback implements MqttCallback {
  
      @Override
      public void connectionLost(Throwable thrwbl) {
          System.out.println(SDF.format(new Date()) + " - MQTT - Perte de connexion... Tentative de reconnexion...");
          connect();
      }
  
      @Override
      public void messageArrived(String topic, MqttMessage message) throws Exception {
          new Thread(new Runnable() {
  
              @Override
              public void run() {
                  String[] tab = topic.split("/", 2);
                  String dbName = tab[0];
                  String collName = (tab.length &gt; 1) ? tab[1] : tab[0];
                  MongoDatabase db = mongoClient.getDatabase(dbName);
                  MongoCollection collection = db.getCollection(collName);
                  Document doc = (Document) collection.find().sort(Sorts.descending("date")).first();
                  Object value = (doc != null) ? doc.get("value") : null;
                  Type type = checkType(message.toString());
                  switch (type) {
                      case Double :
                          if (doc == null || !((Double) value).equals(Double.valueOf(message.toString()))) {
                              collection.insertOne(new Document()
                                  .append("date", SDF.format(new Date()))
                                  .append("value", Double.valueOf(message.toString())));
                          }
                          break;
                      case Boolean :
                          if (doc == null || !((Boolean) value).equals(Boolean.valueOf(message.toString()))) {
                              collection.insertOne(new Document()
                                  .append("date", SDF.format(new Date()))
                                  .append("value", Boolean.valueOf(message.toString())));
                          }
                          break;
                      case String :
                          if (doc == null || !((String) value).equals(message.toString())) {
                              collection.insertOne(new Document()
                                  .append("date", SDF.format(new Date()))
                                  .append("value", message.toString()));
                          }
                          break;
                  }
              }
  
              private Type checkType(String str) {
                  if (str.equalsIgnoreCase("true") || str.equalsIgnoreCase("false"))
                      return Type.Boolean;
                  else {
                      try {
                          Double.valueOf(str);
                          return Type.Double;
                      } catch (NumberFormatException e) {
                          return Type.String;
                      }
                  }
              }
  
          }).start();
      }
  
      @Override
      public void deliveryComplete(IMqttDeliveryToken imdt) {
          throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
      }
  
  }
  </code></pre>
  
  <p>Thanks</p>