database engine initially works

bin/measurementCollector.props

@@ -1,7 +1,7 @@
-;mqtt.broker = tcp://172.16.2.15:1883
+mqtt.broker = tcp://172.16.2.15:1883
-mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
+;mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
-mqtt.username = tron
+;mqtt.username = tron
-mqtt.password = geheim123
+;mqtt.password = geheim123
 
 mbus.dataparser.1 = light,Light,de.hottis.measurementCollector.FinderOnePhasePowerMeter
 mbus.dataparser.2 = computer,Computer,de.hottis.measurementCollector.FinderOnePhasePowerMeter

bin/measurementDataEngine.props

@@ -1,4 +1,8 @@
 db.period = 3600
+db.url = jdbc:mysql://localhost/smarthome
+db.driver = com.mysql.jdbc.Driver
+db.user = smarthome
+db.password = smarthome123
 
 jms.broker = tcp://localhost:61616
 jms.clientid = mdb

src/de/hottis/measurementCollector/DatabaseEngine.java

@@ -1,10 +1,16 @@
 package de.hottis.measurementCollector;
 
 import java.sql.Connection;
+import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.Collection;
+import java.util.HashMap;
 import java.util.Properties;
 import java.util.Timer;
+import java.util.stream.Collectors;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -12,6 +18,9 @@ import org.apache.logging.log4j.Logger;
 public class DatabaseEngine extends Thread implements ITriggerable {
 	static final String DATABASE_ENGINE_PERIOD_PROP = "db.period";
 	static final String DATABASE_URL_PROP = "db.url";
+	static final String DATABASE_USER_PROP = "db.user";
+	static final String DATABASE_PASSWORD_PROP = "db.password";
+	static final String DATABASE_DRIVER_PROP = "db.driver";
 	
 	final protected Logger logger = LogManager.getRootLogger();
 
@@ -22,6 +31,8 @@ public class DatabaseEngine extends Thread implements ITriggerable {
 	private Timer timer;
 	private int period;
 	private String dbUrl;
+	private String dbUsername;
+	private String dbPassword;
 
 	public DatabaseEngine(Properties config, MyQueue<ADataObject> queue) {
 		super("MeasurementCollector.DatabaseEngine");
@@ -31,6 +42,8 @@ public class DatabaseEngine extends Thread implements ITriggerable {
 		this.triggerFlag = false;
 		this.period = Integer.parseInt(this.config.getProperty(DATABASE_ENGINE_PERIOD_PROP));
 		this.dbUrl = this.config.getProperty(DATABASE_URL_PROP);
+		this.dbUsername = this.config.getProperty(DATABASE_USER_PROP);
+		this.dbPassword = this.config.getProperty(DATABASE_PASSWORD_PROP);
 	}
 
 	public void requestShutdown() {
@@ -53,12 +66,48 @@ public class DatabaseEngine extends Thread implements ITriggerable {
 	public void init() throws MeasurementCollectorException {
 		timer = new Timer("DatabaseEngineTrigger");
 		timer.schedule(new TriggerTimer(this), 0, period * 1000);
+		try {
+			Class.forName(config.getProperty(DATABASE_DRIVER_PROP));
+		} catch (ClassNotFoundException e) {
+			logger.error("Database driver class not found", e);
+			throw new MeasurementCollectorException("Database driver class not found", e);
+		}
 	}
 	
 	private String createStatementFromDataObject(ADataObject ado) {
-		return "";
+		StringBuffer sb = new StringBuffer();
+		sb.append("INSERT INTO " + ado.getTableName());
+		sb.append("(name,ts,");
+		sb.append(String.join(",", ado.getValues().keySet()));
+		sb.append(") ");
+		sb.append("VALUES(?,?,");
+		
+		String[] marks = (String[]) ado.getValues().values().stream()
+                .map(c -> "?")
+                .toArray(String[]::new);
+		sb.append(String.join(",", marks));
+		
+		sb.append(")");
+		return sb.toString();
 	}
 	
+	private void bindValue(PreparedStatement pstmt, int pos, Object o) throws SQLException {
+		if (o instanceof Integer) {
+			pstmt.setInt(pos, (Integer)o);
+		} else if (o instanceof Byte) {
+			pstmt.setByte(pos, (Byte)o);
+		} else if (o instanceof Double) {
+			pstmt.setDouble(pos, (Double)o);
+		} else if (o instanceof String) {
+			pstmt.setString(pos, (String)o);
+		} else if (o instanceof LocalDateTime) {
+			pstmt.setTimestamp(pos, Timestamp.valueOf(((LocalDateTime)o)));
+		} else {
+			throw new SQLException("illegal parameter type: " + o.getClass().getName());
+		}
+	}
+	
+	
 	@Override
 	public synchronized void run() {
 		while (! stop) {
@@ -73,46 +122,62 @@ public class DatabaseEngine extends Thread implements ITriggerable {
 			logger.debug("DatabaseEngine has received trigger");
 
 			Connection dbCon = null;
+			HashMap<String, PreparedStatement> preparedStatements = new HashMap<String, PreparedStatement>();
 			try {
 				int itemCnt = 0;
 				
 				while (true) {
-					PreparedStatement pstmt = null;
 					try {
 						ADataObject ado = queue.dequeue();
 						if (ado == null) {
 							logger.warn("DatabaseEngine found no data");
 							break;
 						}
-						String stmtTxt = createStatementFromDataObject(ado);
+						dbCon = DriverManager.getConnection(dbUrl, dbUsername, dbPassword);
-						pstmt = dbCon.prepareStatement(stmtTxt);
-						for (Object o : ado.getValues().values()) {
 
+						String key = ado.getClass().getName();
+						PreparedStatement pstmt;
+						if (! preparedStatements.containsKey(key)) {
+							pstmt = dbCon.prepareStatement(createStatementFromDataObject(ado));
+							preparedStatements.put(key, pstmt);
+						} else {
+							pstmt = preparedStatements.get(key);
+						}
+						bindValue(pstmt, 1, ado.getName());
+						bindValue(pstmt, 2, ado.getTimestamp());
+						int pos = 3;
+						for (Object o : ado.getValues().values()) {
+							bindValue(pstmt, pos, o);
+							pos++;
 						}
 						itemCnt++;
 						logger.info("DatabaseEngine received (" + itemCnt + ") " + ado);
+						logger.info("Statement is " + pstmt.toString());
+						pstmt.execute();
+						logger.info("Database insert executed");
+						pstmt.clearParameters();
 					} catch (SQLException e) {
 						logger.error("SQLException in inner database engine loop", e);
-					} finally {
-						if (pstmt != null) {
-							try {
-								pstmt.close();
-								pstmt = null;
-							} catch (SQLException e) {
-								logger.warn("SQLException when closing statement, nothing will be done");
-							}
-						}
 					}
 				}
 			} catch (Exception e) {
 				logger.error("Exception in outer database engine loop", e);
 			} finally {
-				if (dbCon ! null) {
+				for (PreparedStatement p : preparedStatements.values()) {
+					try {
+						p.close();
+					} catch (SQLException e) {
+						logger.warn("SQLException when closing prepared statement, nothing will be done");
+					}
+				}
+				preparedStatements.clear();
+				if (dbCon != null) {
 					try {
 						dbCon.close();
-						dbCon = null;
 					} catch (SQLException e) {
 						logger.warn("SQLException when closing connection, nothing will be done");
+					} finally {
+						dbCon = null;
 					}
 				}
 			}

src/measurementCollector.props

@@ -1,7 +1,7 @@
-;mqtt.broker = tcp://172.16.2.15:1883
+mqtt.broker = tcp://172.16.2.15:1883
-mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
+;mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
-mqtt.username = tron
+;mqtt.username = tron
-mqtt.password = geheim123
+;mqtt.password = geheim123
 
 mbus.dataparser.1 = light,Light,de.hottis.measurementCollector.FinderOnePhasePowerMeter
 mbus.dataparser.2 = computer,Computer,de.hottis.measurementCollector.FinderOnePhasePowerMeter

src/measurementDataEngine.props

@@ -1,4 +1,8 @@
 db.period = 3600
+db.url = jdbc:mysql://localhost/smarthome
+db.driver = com.mysql.jdbc.Driver
+db.user = smarthome
+db.password = smarthome123
 
 jms.broker = tcp://localhost:61616
 jms.clientid = mdb