wn/MeasurementCollector
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

switch to maven

Browse Source
This commit is contained in:
Wolfgang Hottgenroth
2017-11-22 17:47:10 +01:00
parent 883365227b
commit 132aaf0535
65 changed files with 136 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
src/de/hottis/measurementCollector/ADataObject.java
View File

@ -1,53 +0,0 @@
package de.hottis.measurementCollector;
import java.io.Serializable;
import java.time.LocalDateTime;
import java.util.Map;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
public abstract class ADataObject implements Serializable {
private static final long serialVersionUID = 1L;
final protected Logger logger = LogManager.getRootLogger();
private LocalDateTime timestamp;
private String name;
private Map<String, Object> values;
public ADataObject(LocalDateTime timestamp, String name) {
this.timestamp = timestamp;
this.name = name;
}
public void setValues(Map<String, Object> values) {
this.values = values;
}
abstract public String getTableName();
public LocalDateTime getTimestamp() {
return timestamp;
}
public Map<String, Object> getValues() {
return values;
}
public String getName() {
return name;
}
public String toString() {
StringBuffer sb = new StringBuffer();
sb.append("<");
sb.append(name);
sb.append(", ");
sb.append(timestamp);
sb.append(", ");
sb.append(values.toString());
sb.append(">");
return sb.toString();
}
}

185
src/de/hottis/measurementCollector/DatabaseEngine.java
View File

@ -1,185 +0,0 @@
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.HashMap;
import java.util.Properties;
import java.util.Timer;
import org.apache.logging.log4j.LogManager;
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();
private Properties config;
private MyQueue<ADataObject> queue;
private boolean stop;
private boolean triggerFlag;
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");
this.config = config;
this.queue = queue;
this.stop = false;
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() {
logger.info("Shutdown of database engine requested");
this.stop = true;
try {
this.join();
logger.info("Database engine is down");
} catch (InterruptedException e) {
logger.error("Waiting for shutdown of database engine interrupted");
}
}
public synchronized void trigger() {
logger.debug("DatabaseEngine triggered");
triggerFlag = true;
notify();
}
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) {
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) {
logger.debug("DatabaseEngine is about to wait for (regularly) " + period + "s");
while (! triggerFlag) {
try {
wait();
} catch (InterruptedException e) {
}
}
triggerFlag = false;
logger.debug("DatabaseEngine has received trigger");
Connection dbCon = null;
HashMap<String, PreparedStatement> preparedStatements = new HashMap<String, PreparedStatement>();
try {
int itemCnt = 0;
while (true) {
try {
ADataObject ado = queue.dequeue();
if (ado == null) {
logger.warn("DatabaseEngine found no data");
break;
}
dbCon = DriverManager.getConnection(dbUrl, dbUsername, dbPassword);
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);
}
}
} catch (Exception e) {
logger.error("Exception in outer database engine loop", e);
} finally {
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();
} catch (SQLException e) {
logger.warn("SQLException when closing connection, nothing will be done");
} finally {
dbCon = null;
}
}
}
}
logger.info("Database engine is terminating");
}
}

24
src/de/hottis/measurementCollector/ElectricEnergyDataObject.java
View File

@ -1,24 +0,0 @@
package de.hottis.measurementCollector;
import java.time.LocalDateTime;
import java.util.HashMap;
public class ElectricEnergyDataObject extends ADataObject {
private static final long serialVersionUID = 1L;
static final String ENERGY_KEY = "energy";
static final String POWER_KEY = "power";
static final String TABLE_NAME = "ElectricEnergy";
public ElectricEnergyDataObject(LocalDateTime timestamp, String name, double energy, double power) {
super(timestamp, name);
HashMap<String, Object> values = new HashMap<String, Object>();
values.put(ENERGY_KEY, energy);
values.put(POWER_KEY, power);
setValues(values);
}
@Override
public String getTableName() {
return TABLE_NAME;
}
}

5
src/de/hottis/measurementCollector/ITriggerable.java
View File

@ -1,5 +0,0 @@
package de.hottis.measurementCollector;
public interface ITriggerable {
public void trigger();
}

124
src/de/hottis/measurementCollector/JmsTopic.java
View File

@ -1,124 +0,0 @@
package de.hottis.measurementCollector;
import java.io.Serializable;
import java.util.Properties;
import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
public class JmsTopic<T extends Serializable> implements MyQueue<T> {
static final String JMS_PARSED_DATA_TOPIC = "jms.parseddata.topic";
static final String JMS_CLIENTID_PROP = "jms.clientid";
static final String JMS_BROKER ="jms.broker";
static enum Mode { CONSUMER, PRODUCER, CONSUMER_PRODUCER };
final protected Logger logger = LogManager.getRootLogger();
private String clientId;
private Properties config;
private Connection connection;
private Session session;
private Topic topic;
private MessageProducer producer;
private MessageConsumer consumer;
private String topicTxt;
private String broker;
private Mode mode;
public JmsTopic(Properties config, Mode mode) {
this.config = config;
clientId = this.config.getProperty(JMS_CLIENTID_PROP);
topicTxt = this.config.getProperty(JMS_PARSED_DATA_TOPIC);
broker = this.config.getProperty(JMS_BROKER);
this.mode = mode;
}
public void init() throws MeasurementCollectorException {
try {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(broker);
connectionFactory.setTrustAllPackages(true);
logger.debug("connectionFactory: " + connectionFactory);
connection = connectionFactory.createConnection();
connection.setClientID(clientId);
connection.start();
logger.debug("connection: " + connection);
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
logger.debug("session: " + session);
topic = session.createTopic(topicTxt);
logger.debug("topic: " + topic);
if ((mode == Mode.PRODUCER) || (mode == Mode.CONSUMER_PRODUCER)) {
producer = session.createProducer(topic);
producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
logger.debug("producer: " + producer);
} else {
producer = null;
}
if ((mode == Mode.CONSUMER) || (mode == Mode.CONSUMER_PRODUCER)) {
consumer = session.createDurableSubscriber(topic, "DatabaseEngine");
logger.debug("consumer: " + consumer);
} else {
consumer = null;
}
} catch (JMSException e) {
logger.error("JMSException in AJmsQueue.init", e);
throw new MeasurementCollectorException("JMSException in AJmsTopic.init", e);
}
}
public T dequeue() throws MeasurementCollectorException {
if (consumer == null) {
throw new MeasurementCollectorException("This is no consumer");
}
try {
ObjectMessage message = (ObjectMessage) consumer.receiveNoWait();
T item;
if (message != null) {
@SuppressWarnings("unchecked")
T t = (T) message.getObject();
item = t;
logger.debug("message dequeued");
} else {
item = null;
}
return item;
} catch (JMSException e) {
logger.error("JMSException in JmsTopic.dequeue", e);
logger.error("Calling init");
init();
throw new MeasurementCollectorException("JMSException in JmsTopic.dequeue", e);
}
}
public void enqueue(T item) throws MeasurementCollectorException {
if (producer == null) {
throw new MeasurementCollectorException("This is no producer");
}
try {
ObjectMessage message = session.createObjectMessage();
message.setObject(item);
producer.send(message);
logger.debug("message enqueued");
} catch (JMSException e) {
logger.error("JMSException in JmsTopic.enqueue", e);
logger.error("Calling init");
init();
throw new MeasurementCollectorException("JMSException in JmsTopic.enqueue", e);
}
}
}

13
src/de/hottis/measurementCollector/MeasurementCollectorException.java
View File

@ -1,13 +0,0 @@
package de.hottis.measurementCollector;
@SuppressWarnings("serial")
public class MeasurementCollectorException extends Exception {
public MeasurementCollectorException(String msg, Throwable cause) {
super(msg, cause);
}
public MeasurementCollectorException(String msg) {
super(msg);
}
}

30
src/de/hottis/measurementCollector/MeasurementDatabaseEngine.java
View File

@ -1,30 +0,0 @@
package de.hottis.measurementCollector;
import java.util.Properties;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
public class MeasurementDatabaseEngine {
static final String PROPS_FILENAME = "measurementDataEngine.props";
static final Logger logger = LogManager.getRootLogger();
public static void main(String[] args) throws Exception {
logger.info("MeasurementDatabaseEngine starting");
final Properties config = new Properties();
config.load(MeasurementDatabaseEngine.class.getClassLoader().getResourceAsStream(PROPS_FILENAME));
logger.debug("Configuration loaded");
JmsTopic<ADataObject> queue = new JmsTopic<ADataObject>(config, JmsTopic.Mode.CONSUMER);
queue.init();
DatabaseEngine databaseEngine = new DatabaseEngine(config, queue);
databaseEngine.init();
databaseEngine.start();
logger.debug("DatabaseEngine started");
}
}

8
src/de/hottis/measurementCollector/MyQueue.java
View File

@ -1,8 +0,0 @@
package de.hottis.measurementCollector;
import java.io.Serializable;
public interface MyQueue<T extends Serializable> {
public T dequeue() throws MeasurementCollectorException;
public void enqueue(T item) throws MeasurementCollectorException;
}

21
src/de/hottis/measurementCollector/TemperatureDataObject.java
View File

@ -1,21 +0,0 @@
package de.hottis.measurementCollector;
import java.time.LocalDateTime;
import java.util.HashMap;
public class TemperatureDataObject extends ADataObject {
private static final long serialVersionUID = 1L;
static final String TEMPERATURE_KEY = "temperature";
static final String TABLE_NAME = "Temperature";
public TemperatureDataObject(LocalDateTime timestamp, String name, double temperature) {
super(timestamp, name);
HashMap<String, Object> values = new HashMap<String, Object>();
values.put(TEMPERATURE_KEY, temperature);
setValues(values);
}
public String getTableName() {
return TABLE_NAME;
}
}

15
src/de/hottis/measurementCollector/TriggerTimer.java
View File

@ -1,15 +0,0 @@
package de.hottis.measurementCollector;
import java.util.TimerTask;
public class TriggerTimer extends TimerTask {
private ITriggerable triggerable;
public TriggerTimer(ITriggerable triggerable) {
this.triggerable = triggerable;
}
public void run() {
triggerable.trigger();
}
}

0
src/de/hottis/measurementCollector/ADataParser.java → src/main/java/de/hottis/measurementCollector/ADataParser.java
View File

0
src/de/hottis/measurementCollector/AMessageParser.java → src/main/java/de/hottis/measurementCollector/AMessageParser.java
View File

0
src/de/hottis/measurementCollector/FinderOnePhasePowerMeter.java → src/main/java/de/hottis/measurementCollector/FinderOnePhasePowerMeter.java
View File

0
src/de/hottis/measurementCollector/FinderThreePhasePowerMeter.java → src/main/java/de/hottis/measurementCollector/FinderThreePhasePowerMeter.java
View File

0
src/de/hottis/measurementCollector/HottisFourChannelThermometer.java → src/main/java/de/hottis/measurementCollector/HottisFourChannelThermometer.java
View File

0
src/de/hottis/measurementCollector/MBusParser.java → src/main/java/de/hottis/measurementCollector/MBusParser.java
View File

0
src/de/hottis/measurementCollector/MeasurementCollector.java → src/main/java/de/hottis/measurementCollector/MeasurementCollector.java
View File

0
src/de/hottis/measurementCollector/MqttReceiver.java → src/main/java/de/hottis/measurementCollector/MqttReceiver.java
View File

0
src/log4j2.xml → src/main/resources/log4j2.xml
View File

17
src/measurementCollector.props
View File

@ -1,17 +0,0 @@
mqtt.broker = tcp://172.16.2.15:1883
;mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
;mqtt.username = tron
;mqtt.password = geheim123
mbus.dataparser.1 = light,Light,de.hottis.measurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.2 = computer,Computer,de.hottis.measurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.3 = laundry,Laundry,de.hottis.measurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.4 = dryer,Dryer,de.hottis.measurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.5 = dishwasher,Dishwasher,de.hottis.measurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.6 = freezer,Freezer,de.hottis.measurementCollector.FinderOnePhasePowerMeter
mbus.dataparser.7 = electricity,Total,de.hottis.measurementCollector.FinderThreePhasePowerMeter
mbus.dataparser.8 = thermom.,null,de.hottis.measurementCollector.HottisFourChannelThermometer
jms.broker = tcp://localhost:61616
jms.clientid = mcol
jms.parseddata.topic = IoT/Measurement

9
src/measurementDataEngine.props
View File

@ -1,9 +0,0 @@
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
jms.parseddata.topic = IoT/Measurement

38
src/test/java/de/hottis/measurementCollector/AppTest.java Normal file
View File

@ -0,0 +1,38 @@
package de.hottis.measurementCollector;
import junit.framework.Test;
import junit.framework.TestCase;
import junit.framework.TestSuite;
/**
* Unit test for simple App.
*/
public class AppTest
extends TestCase
{
/**
* Create the test case
*
* @param testName name of the test case
*/
public AppTest( String testName )
{
super( testName );
}
/**
* @return the suite of tests being tested
*/
public static Test suite()
{
return new TestSuite( AppTest.class );
}
/**
* Rigourous Test :-)
*/
public void testApp()
{
assertTrue( true );
}
}