001package var.mom.mqtt.smarthome; 002 003import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; 004import org.eclipse.paho.client.mqttv3.MqttCallback; 005import org.eclipse.paho.client.mqttv3.MqttClient; 006import org.eclipse.paho.client.mqttv3.MqttException; 007import org.eclipse.paho.client.mqttv3.MqttMessage; 008 009/** 010 * client for the Smart Home application: prints out alarm messages to the 011 * console if trigger condition in topic is detected 012 * 013 * @author Sandro Leuchter 014 * 015 */ 016class AlarmSubscriber { 017 /** 018 * trigger alarm if observation > threshold 019 */ 020 static double threshold; 021 022 /** 023 * main routine and starting point of program: subscribes to topic 024 * <tt>Conf.TOPICSTART/args[0]/args[1]/args[2]</tt> i.e. a specific sensor and 025 * prints messages to console if observation > threshold 026 * 027 * @param args[0] floor where sensor is located 028 * @param args[1] room where sensor is located 029 * @param args[2] type of sensor 030 * @param args[3] threshold 031 * @throws MqttException Paho library exceptions that have something to do with MQTT 032 */ 033 public static void main(String[] args) throws MqttException { 034 String topic = Conf.TOPICSTART + "/" + args[0] + "/" + args[1] + "/" + args[2]; 035 threshold = Double.valueOf(args[3]); 036 MqttClient client = new MqttClient(Conf.BROKER, MqttClient.generateClientId()); 037 client.setCallback(new MqttCallback() { 038 @Override 039 public void messageArrived(String topic, MqttMessage m) throws Exception { 040 try { 041 double observation = Double.valueOf(m.toString()); 042 if (observation > threshold) { 043 System.out.println("ALARM: " + topic + ": " + m); 044 } 045 } catch (NumberFormatException e) { 046 System.out.println(m); // message is not a sensor 047 // observation: will be printed 048 // instead 049 } 050 } 051 052 @Override 053 public void deliveryComplete(IMqttDeliveryToken arg0) { 054 } 055 056 @Override 057 public void connectionLost(Throwable arg0) { 058 } 059 }); 060 client.connect(); 061 client.subscribe(topic); 062 try { 063 while (true) { 064 Thread.sleep(1000); 065 } 066 } catch (InterruptedException e) { 067 System.err.println(e.getMessage()); 068 } finally { 069 try { 070 client.disconnect(); 071 } catch (MqttException e) { 072 // unrecoverable 073 } 074 } 075 } 076}