initial

.classpath

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<classpath>
+	<classpathentry kind="src" path="src"/>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/JavaSE-1.8"/>
+	<classpathentry kind="lib" path="/home/wn/workspace-java/MeasurementCollector/libraries/jmbus-3.0.1.jar"/>
+	<classpathentry kind="lib" path="/home/wn/workspace-java/MeasurementCollector/libraries/org.eclipse.paho.client.mqttv3-1.2.0.jar"/>
+	<classpathentry kind="output" path="bin"/>
+</classpath>

.project

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<projectDescription>
+	<name>MeasurementCollector</name>
+	<comment></comment>
+	<projects>
+	</projects>
+	<buildSpec>
+		<buildCommand>
+			<name>org.eclipse.jdt.core.javabuilder</name>
+			<arguments>
+			</arguments>
+		</buildCommand>
+	</buildSpec>
+	<natures>
+		<nature>org.eclipse.jdt.core.javanature</nature>
+	</natures>
+</projectDescription>

.settings/org.eclipse.jdt.core.prefs

@@ -0,0 +1,11 @@
+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.inlineJsrBytecode=enabled
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.8
+org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
+org.eclipse.jdt.core.compiler.compliance=1.8
+org.eclipse.jdt.core.compiler.debug.lineNumber=generate
+org.eclipse.jdt.core.compiler.debug.localVariable=generate
+org.eclipse.jdt.core.compiler.debug.sourceFile=generate
+org.eclipse.jdt.core.compiler.problem.assertIdentifier=error
+org.eclipse.jdt.core.compiler.problem.enumIdentifier=error
+org.eclipse.jdt.core.compiler.source=1.8

bin/measurementCollector.props

@@ -0,0 +1,3 @@
+mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
+mqtt.username = tron
+mqtt.password = geheim123

src/de/hottis/MeasurementCollector/AMessageParser.java

@@ -0,0 +1,17 @@
+package de.hottis.MeasurementCollector;
+
+import java.time.LocalDateTime;
+
+public abstract class AMessageParser {
+	private String topic;
+	
+	public AMessageParser(String topic) {
+		this.topic = topic;
+	}
+	
+	public String getTopic() {
+		return this.topic;
+	}
+	
+	abstract public void execute(LocalDateTime timestamp, String msgPayload);
+}

src/de/hottis/MeasurementCollector/MeasurementCollector.java

@@ -0,0 +1,51 @@
+package de.hottis.MeasurementCollector;
+
+import java.util.Properties;
+
+public class MeasurementCollector {
+	static final String PROPS_FILENAME = "measurementCollector.props";
+	
+	
+	public static void main(String[] args) throws Exception {
+		System.out.println("MeasurementCollector starting");
+		
+		final Properties config = new Properties();
+		config.load(MeasurementCollector.class.getClassLoader().getResourceAsStream(PROPS_FILENAME));
+		
+		MqttReceiver mqttReceiver = new MqttReceiver(config);
+		mqttReceiver.connect();
+
+		TestParser testParser = new TestParser();
+		mqttReceiver.registerParser(testParser);
+		
+	}
+		
+/*		
+import org.openmuc.jmbus.DataRecord;
+import org.openmuc.jmbus.MBusMessage;
+import org.openmuc.jmbus.VariableDataStructure;
+import java.util.List;
+
+		// String txtMsg = "68 38 38 68 08 54 72 21 00 13 00 2E 19 24 02 43 00 00 00 8C 10 04 97 84 16 00 8C 11 04 97 84 16 00 02 FD C9 FF 01 DF 00 02 FD DB FF 01 0D 00 02 AC FF 01 1B 00 82 40 AC FF 01 00 00 1A 16";
+		String txtMsg = "68 61 61 68 08 21 72 00 00 00 00 00 00 01 00 B5 00 00 00 01 24 08 01 25 10 01 26 0D 02 27 01 01 05 67 C8 44 C0 3C 05 67 34 E9 0C 41 05 67 B7 F3 9A 41 05 67 71 86 25 41 0F E0 7A 32 00 B5 99 04 00 73 98 02 00 00 00 00 00 A6 06 00 00 A2 C3 7F 3F A5 BA 7F 3F 85 A7 7F 3F E7 F9 7F 3F CD CC CC 3D E8 03 00 00 12 16";
+		String [] octetsTxt = txtMsg.split(" ");
+		byte [] octets = new byte[octetsTxt.length];
+		System.out.println("Start");
+		for (int i = 0; i < octetsTxt.length; i++) {
+			System.out.println(octetsTxt[i]);
+			octets[i] = (byte)(Integer.parseInt(octetsTxt[i], 16) & 0xff);
+		}
+		System.out.println("End");
+		
+		MBusMessage mbusMsg = MBusMessage.decode(octets, octets.length);
+		//System.out.println("MBusMessage: " + mbusMsg.toString());
+		VariableDataStructure variableDataStructure = mbusMsg.getVariableDataResponse();
+		variableDataStructure.decode();
+		List<DataRecord> dataRecords = variableDataStructure.getDataRecords();
+		
+		for (DataRecord dataRecord : dataRecords) {
+			System.out.println(dataRecord.getScaledDataValue() + " " + dataRecord.getUnit().getUnit());
+		}
+	}
+*/
+}

src/de/hottis/MeasurementCollector/MeasurementCollectorException.java

@@ -0,0 +1,8 @@
+package de.hottis.MeasurementCollector;
+
+public class MeasurementCollectorException extends Exception {
+	public MeasurementCollectorException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+
+}

src/de/hottis/MeasurementCollector/MqttReceiver.java

@@ -0,0 +1,72 @@
+package de.hottis.MeasurementCollector;
+
+import java.time.LocalDateTime;
+import java.util.HashMap;
+import java.util.Properties;
+
+import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+
+
+public class MqttReceiver {
+	static final String MQTT_BROKER_PROP = "mqtt.broker";
+	static final String MQTT_CLIENTID_PROP = "mqtt.clientid";
+	static final String MQTT_USERNAME_PROP = "mqtt.username";
+	static final String MQTT_PASSWORD_PROP = "mqtt.password";
+	
+	
+	class Listener implements IMqttMessageListener {
+		public void messageArrived(String topic, MqttMessage payload) {
+			parsers.get(topic).execute(LocalDateTime.now(), payload.toString());
+		}
+	}
+	
+	private final String broker;
+	private final String clientId;
+	private final MqttConnectOptions connOpts;
+	private MqttClient client;
+	private HashMap<String, AMessageParser> parsers;
+	private final Listener listener = new Listener();
+	
+	public MqttReceiver(Properties config) {
+		broker = config.getProperty(MQTT_BROKER_PROP, "localhost");
+		clientId = config.getProperty(MQTT_CLIENTID_PROP, "MeasurementCollector.MqttReceiver");
+		connOpts = new MqttConnectOptions();
+		String username = config.getProperty(MQTT_USERNAME_PROP);
+		String password = config.getProperty(MQTT_PASSWORD_PROP);
+		if (username != null && password != null) {
+			connOpts.setUserName(username);
+			connOpts.setPassword(password.toCharArray());
+		}
+		
+		parsers = new HashMap<String, AMessageParser>();
+	}
+	
+	public void connect() throws MeasurementCollectorException {
+		try {
+			client = new MqttClient(broker, clientId);
+			client.connect(connOpts);
+			while (! client.isConnected()) {
+				System.out.print(".");
+			}
+			System.out.println();
+			System.out.println("Connected");
+		} catch (MqttException e) {
+			throw new MeasurementCollectorException("MqttReceiver.connect", e);
+		}
+	}
+	
+	public void registerParser(AMessageParser parser) throws MeasurementCollectorException {
+		try {
+			parsers.put(parser.getTopic(), parser);
+			client.subscribe(parser.getTopic(), listener);
+			System.out.println("Subscribed: " + parser.getTopic());
+		} catch (MqttException e) {
+			throw new MeasurementCollectorException("MqttReceiver.registerParser", e);
+		}
+	}
+}

src/de/hottis/MeasurementCollector/TestParser.java

@@ -0,0 +1,16 @@
+package de.hottis.MeasurementCollector;
+
+import java.time.LocalDateTime;
+
+public class TestParser extends AMessageParser {
+
+	public TestParser() {
+		super("IoT/Watchdog");
+	}
+
+	@Override
+	public void execute(LocalDateTime timestamp, String msgPayload) {
+		System.out.println(timestamp.toString() + " " + msgPayload);
+	}
+
+}

src/measurementCollector.props

@@ -0,0 +1,3 @@
+mqtt.broker = tcp://eupenstrasse20.dynamic.hottis.de:2883
+mqtt.username = tron
+mqtt.password = geheim123