fixing power messages, remove semaphore, allow qos and retain configuration

Author: magcode

Diff for: README.md

@@ -14,6 +14,8 @@ rootTopic=home                    # the mqtt root topic
 refresh=60                        # number of seconds for MQTT status updates. Do not go below 60!
 mqttServer=tcp://192.168.0.1      # IP or hostname of your mqtt broker
 logLevel=INFO                     # logLevel
+retained=true                     # MQTT messages will be sent retained. Defaults to "false"
+qos=1										# MQTT messages will be sent with qos=1. Defaults to "0"
 
 daikin1.host=192.168.0.2          # IP adress of your first Daikin Wifi adapter
 daikin1.name=ac-room1             # a name for the Daikin device, used in the MQTT topic
@@ -116,8 +118,7 @@ home/ac-room1/aircon/mode/set
 
 Notes
 * `otemp` will go to `0` in case the device is not cooling or heating.
-* You actually don't need to use `power` to turn the AC on or off. Sending `mode=Cool` or `mode=None` does the job as well.
-* Possible values for `mode`, `fandirection` etc. can be found [here](https://bitbucket.org/m-do/jdaikin/src/default/src/main/java/net/jonathangiles/daikin/enums/) 
+* Possible values for `mode`, `fandirection` etc. can be found [here](https://github.com/magcode/daikin-mqtt/tree/master/src/main/java/org/magcode/daikin/connector/enums) 
 
 # Openhab integration example
 
@@ -129,7 +130,7 @@ Thing mqtt:topic:ac-room1 "Aircondition room 1" (mqtt:broker:mosquitto) {
         Type number : intemp "Inside temp" [ stateTopic="home/ac-room1/aircon/intemp"] 
         Type number : outtemp "Outside temp" [ stateTopic="home/ac-room1/aircon/otemp"] 
         Type number : targettemp "Target temp" [ stateTopic="home/ac-room1/aircon/targettemp"] 
-        Type switch : power "Power" [ stateTopic="home/ac-room1/aircon/power", commandTopic="home/ac-room1/aircon/power/set", on="1", off="0"]
+        Type switch : power "Power" [ stateTopic="home/ac-room1/aircon/power", commandTopic="home/ac-room1/aircon/power/set", on="On", off="Off"]
         Type switch : online "Online" [ stateTopic="home/ac-room1/$state", on="ready", off="lost"]
 }
 ```

Diff for: src/main/java/org/magcode/daikin/DaikinMqttClient.java

@@ -31,8 +31,6 @@
 import org.magcode.daikin.mqtt.DevicePublisher;
 import org.magcode.daikin.mqtt.NodePublisher;
 
-import java.util.concurrent.Semaphore;
-
 /**
  * @author magcode MQTT Gateway for Daikin Wifi Adapter
  *
@@ -42,16 +40,16 @@ public class DaikinMqttClient {
 	private static String mqttServer;
 	private static int refresh = 60;
 	private static String rootTopic;
+	private static boolean retained;
+	private static int qos;
 	private static MqttClient mqttClient;
 	public static final String nodeName = "aircon";
 	private static final int deviceRefresh = 600;
-	private static final int MAX_INFLIGHT = 100;
-	private static Semaphore semaphore;
+	private static final int MAX_INFLIGHT = 200;
 	private static Logger logger = LogManager.getLogger(DaikinMqttClient.class);
 	private static String logLevel = "INFO";
 
 	public static void main(String[] args) throws Exception {
-		semaphore = new Semaphore(MAX_INFLIGHT);
 		daikins = new HashMap<String, DaikinConfig>();
 		readProps();
 		reConfigureLogger();
@@ -66,12 +64,12 @@ public static void main(String[] args) throws Exception {
 
 		// start mqtt node publisher
 		ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
-		Runnable nodePublisher = new NodePublisher(daikins, rootTopic, mqttClient, semaphore);
+		Runnable nodePublisher = new NodePublisher(daikins, rootTopic, mqttClient, retained, qos);
 		ScheduledFuture<?> nodePublisherFuture = executor.scheduleAtFixedRate(nodePublisher, 10, refresh,
 				TimeUnit.SECONDS);
 
 		// start mqtt device publisher
-		Runnable devicePublisher = new DevicePublisher(daikins, rootTopic, mqttClient, deviceRefresh, semaphore);
+		Runnable devicePublisher = new DevicePublisher(daikins, rootTopic, mqttClient, deviceRefresh, retained, qos);
 		ScheduledFuture<?> devicePublisherFuture = executor.scheduleAtFixedRate(devicePublisher, 15, deviceRefresh,
 				TimeUnit.SECONDS);
 
@@ -120,6 +118,8 @@ private static void readProps() {
 			refresh = Integer.parseInt(props.getProperty("refresh", "60"));
 			mqttServer = props.getProperty("mqttServer", "tcp://localhost");
 			logLevel = props.getProperty("logLevel", "INFO");
+			retained = Boolean.valueOf(props.getProperty("retained", "false"));
+			qos = Integer.valueOf(props.getProperty("qos", "0"));
 			Enumeration<?> e = props.propertyNames();
 
 			while (e.hasMoreElements()) {
@@ -160,12 +160,17 @@ private static void startMQTTClient() throws MqttException {
 		mqttClient = new MqttClient(mqttServer, "client-for-daikin-on-" + hostName, new MemoryPersistence());
 		MqttConnectOptions connOpt = new MqttConnectOptions();
 		connOpt.setCleanSession(true);
-		connOpt.setKeepAliveInterval(30);
 		connOpt.setMaxInflight(MAX_INFLIGHT);
 		connOpt.setAutomaticReconnect(true);
-		mqttClient.setCallback(new Subscriber(daikins, rootTopic, semaphore));
+		mqttClient.setCallback(new Subscriber(daikins, rootTopic));
 		mqttClient.connect();
 		logger.info("Connected to MQTT broker.");
+		try {
+			// give some time before subscribing
+			Thread.sleep(200);
+		} catch (InterruptedException e) {
+			//
+		}
 		for (Entry<String, DaikinConfig> entry : daikins.entrySet()) {
 			DaikinConfig value = entry.getValue();
 			String subTopic = rootTopic + "/" + value.getName() + "/aircon/+/set";

Diff for: src/main/java/org/magcode/daikin/connector/enums/Power.java

@@ -28,4 +28,13 @@ public String getValue() {
 	public static Power get(String value) {
 		return lookup.get(value);
 	}
+
+	public static Power getFromString(String power) {
+		try {
+			return Power.valueOf(power);
+		} catch (Exception e) {
+			// ignore
+		}
+		return Power.Off;
+	}
 }

Diff for: src/main/java/org/magcode/daikin/mqtt/DevicePublisher.java

@@ -4,7 +4,6 @@
 import java.net.UnknownHostException;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.Semaphore;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
@@ -20,12 +19,13 @@ public class DevicePublisher extends Publisher implements Runnable {
 	private static Logger logger = LogManager.getLogger(DevicePublisher.class);
 
 	public DevicePublisher(Map<String, DaikinConfig> daikinHosts, String topic, MqttClient mqttClient,
-			int devicerefresh, Semaphore semaphore) {
+			int devicerefresh,boolean retained, int qos) {
 		this.daikinHosts = daikinHosts;
 		this.topic = topic;
 		this.mqttClient = mqttClient;
 		this.deviceRefresh = devicerefresh;
-		this.semaphore = semaphore;
+		this.qos = qos;
+		this.retained = retained;
 	}
 
 	@Override

Diff for: src/main/java/org/magcode/daikin/mqtt/NodePublisher.java

@@ -3,7 +3,6 @@
 import java.util.Date;
 import java.util.Map;
 import java.util.Map.Entry;
-import java.util.concurrent.Semaphore;
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -23,12 +22,13 @@ public class NodePublisher extends Publisher implements Runnable {
 	private Map<String, DaikinConfig> daikinHosts;
 	private static Logger logger = LogManager.getLogger(NodePublisher.class);
 
-	public NodePublisher(Map<String, DaikinConfig> daikinHosts, String topic, MqttClient mqttClient,
-			Semaphore semaphore) {
+	public NodePublisher(Map<String, DaikinConfig> daikinHosts, String topic, MqttClient mqttClient, boolean retained,
+			int qos) {
 		this.daikinHosts = daikinHosts;
 		this.topic = topic;
 		this.mqttClient = mqttClient;
-		this.semaphore = semaphore;
+		this.qos = qos;
+		this.retained = retained;
 	}
 
 	@Override
@@ -52,15 +52,8 @@ public void run() {
 				long uptime = (now.getTime() - daikinConfig.getOnlineSince().getTime()) / 1000;
 				Publish(deviceTopic + "/$stats/uptime", "" + uptime);
 
-				Publish(nodeTopic + TopicConstants.PR_POWER, status.getPower().getValue());
-
-				// we force "none" if the device is OFF
-				if (status.getPower() == Power.On) {
-					Publish(nodeTopic + TopicConstants.PR_MODE, status.getMode().toString());
-				} else {
-					Publish(nodeTopic + TopicConstants.PR_MODE, Mode.None.toString());
-				}
-
+				Publish(nodeTopic + TopicConstants.PR_POWER, status.getPower().toString());
+				Publish(nodeTopic + TopicConstants.PR_MODE, status.getMode().toString());
 				Publish(nodeTopic + TopicConstants.PR_TARGETTEMP, status.getTargetTemp());
 				Publish(nodeTopic + TopicConstants.PR_INTEMP, status.getInsideTemp());
 				Publish(nodeTopic + TopicConstants.PR_OUTTEMP, status.getOutsideTemp());
@@ -74,7 +67,7 @@ public void run() {
 				Publish(deviceTopic + "/$stats/uptime", 0);
 				Publish(deviceTopic + "/$state", "lost");
 				Publish(nodeTopic + TopicConstants.PR_MODE, Mode.None.toString());
-				Publish(nodeTopic + TopicConstants.PR_POWER, false);
+				Publish(nodeTopic + TopicConstants.PR_POWER, Power.Off.toString());
 				Publish(nodeTopic + TopicConstants.PR_OUTTEMP, 0.0f);
 				Publish(nodeTopic + TopicConstants.PR_INTEMP, 0.0f);
 				Publish(nodeTopic + TopicConstants.PR_TARGETTEMP, 0.0f);

Diff for: src/main/java/org/magcode/daikin/mqtt/Publisher.java

@@ -1,7 +1,5 @@
 package org.magcode.daikin.mqtt;
 
-import java.util.concurrent.Semaphore;
-
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.eclipse.paho.client.mqttv3.MqttClient;
@@ -11,7 +9,8 @@
 
 public abstract class Publisher {
 	protected MqttClient mqttClient;
-	protected Semaphore semaphore;
+	protected int qos = 0;
+	protected boolean retained = false;
 	private static Logger logger = LogManager.getLogger(Publisher.class);
 
 	protected void Publish(String topic, String payload) {
@@ -33,14 +32,9 @@ protected void Publish(String topic, int payload) {
 
 	private void Publish(String topic, MqttMessage message) {
 		try {
-			message.setRetained(true);
-			if (semaphore.availablePermits() < 10) {
-				logger.warn("Semaphore available permits: {}", semaphore.availablePermits());
-			}
-			semaphore.acquire();
+			message.setRetained(this.retained);
+			message.setQos(this.qos);
 			this.mqttClient.publish(topic, message);
-		} catch (InterruptedException e) {
-			logger.error("InterruptedException", e);
 		} catch (MqttPersistenceException e) {
 			logger.error("MqttPersistenceException", e);
 		} catch (MqttException e) {

Diff for: src/main/java/org/magcode/daikin/mqtt/Subscriber.java

@@ -1,7 +1,6 @@
 package org.magcode.daikin.mqtt;
 
 import java.util.Map;
-import java.util.concurrent.Semaphore;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
@@ -22,12 +21,10 @@ public class Subscriber implements MqttCallback {
 	private Map<String, DaikinConfig> daikins;
 	private String rootTopic;
 	private static Logger logger = LogManager.getLogger(Subscriber.class);
-	private Semaphore semaphore;
 
-	public Subscriber(Map<String, DaikinConfig> daikins, String topic, Semaphore semaphore) {
+	public Subscriber(Map<String, DaikinConfig> daikins, String topic) {
 		this.daikins = daikins;
 		this.rootTopic = topic;
-		this.semaphore = semaphore;
 	}
 
 	@Override
@@ -89,7 +86,7 @@ public void messageArrived(String topic, MqttMessage message) throws Exception {
 					break;
 				case TopicConstants.PR_POWER:
 					String powerString = message.toString();
-					Power power = Power.valueOf(powerString);
+					Power power = Power.getFromString(powerString);
 					logger.info("Sending power={} to {}", power, targetDaikin.getName());
 					targetDaikin.getDaikin().updateStatus();
 					targetDaikin.getDaikin().setPower(power);
@@ -117,6 +114,6 @@ public void messageArrived(String topic, MqttMessage message) throws Exception {
 
 	@Override
 	public void deliveryComplete(IMqttDeliveryToken token) {
-		semaphore.release();
+
 	}
 }