aqhome apps: more work on transition to events2.

apps/aqhome-mqttlog/0BUILD

@@ -43,6 +43,7 @@
       server.h
       server_p.h
       s_publish.h
+      s_setdata.h
     </headers>
   
     <sources>
@@ -54,6 +55,7 @@
       c_setdata.c
       server.c
       s_publish.c
+      s_setdata.c
     </sources>
 
     <useTargets>

apps/aqhome-mqttlog/main.c

@@ -191,7 +191,7 @@ void _runService(AQH_OBJECT *aqh, AQH_EVENT_LOOP *eventLoop)
       break;
     }
   } /* while */
-
+  DBG_ERROR(NULL, "Leaving server");
 
   rv=AQH_MqttLogServer_SaveRuntimeDeviceFiles(aqh);
   if (rv<0) {

apps/aqhome-mqttlog/s_publish.c

@@ -214,19 +214,27 @@ void _sendMessage(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *device, const AQ
     DBG_ERROR(NULL, "Invalid value received from MQTT server (%s)", rcvdValue?rcvdValue:"<empty>");
   }
   else {
-    AQH_MESSAGE *pubMsg;
     uint64_t now;
     AQH_VALUE *msgValue;
 
     now=(uint64_t) time(NULL);
     msgValue=_mkMessageValue(device, value);
-    pubMsg=AQH_IpcdMessageMultiData_newForOne(AQH_MSGTYPE_IPC_DATA_UPDATEDATA,
-                                              AQH_Endpoint_GetNextMessageId(xo->brokerEndpoint), 0,
-                                              msgValue, now, f);
-    DBG_INFO(AQH_LOGDOMAIN, "BROKER UPDATE_DATA %s/%s: %f",
-             deviceName?deviceName:"<no device name>",
-             AQH_Value_GetName(msgValue), f);
-    AQH_Endpoint_AddMsgOut(xo->brokerEndpoint, pubMsg);
+    if (xo->brokerEndpoint) {
+      AQH_MESSAGE *pubMsg;
+
+      pubMsg=AQH_IpcdMessageMultiData_newForOne(AQH_MSGTYPE_IPC_DATA_UPDATEDATA,
+                                                AQH_Endpoint_GetNextMessageId(xo->brokerEndpoint), 0,
+                                                msgValue, now, f);
+      DBG_INFO(AQH_LOGDOMAIN, "BROKER UPDATE_DATA %s/%s: %f",
+               deviceName?deviceName:"<no device name>",
+               AQH_Value_GetName(msgValue), f);
+      AQH_Endpoint_AddMsgOut(xo->brokerEndpoint, pubMsg);
+    }
+    else {
+      DBG_INFO(AQH_LOGDOMAIN, "Skipping BROKER UPDATE_DATA %s/%s: %f",
+               deviceName?deviceName:"<no device name>",
+               AQH_Value_GetName(msgValue), f);
+    }
     AQH_Value_free(msgValue);
   }
 }
@@ -269,14 +277,19 @@ void _sendAnnounceValueMessage(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *dev
   AQH_VALUE *msgValue;
 
   msgValue=_mkMessageValue(device, value);
-  pubMsg=AQH_IpcdMessageValues_newForOne(AQH_MSGTYPE_IPC_DATA_ANNOUNCEVALUE,
-                                         AQH_Endpoint_GetNextMessageId(xo->brokerEndpoint), 0,
-                                         0, msgValue);
-  if (pubMsg) {
-    DBG_INFO(AQH_LOGDOMAIN, "BROKER ANNOUNCE_VALUE %s", AQH_Value_GetName(msgValue));
-    AQH_Endpoint_AddMsgOut(xo->brokerEndpoint, pubMsg);
+  if (xo->brokerEndpoint) {
+    pubMsg=AQH_IpcdMessageValues_newForOne(AQH_MSGTYPE_IPC_DATA_ANNOUNCEVALUE,
+                                           AQH_Endpoint_GetNextMessageId(xo->brokerEndpoint), 0,
+                                           0, msgValue);
+    if (pubMsg) {
+      DBG_INFO(AQH_LOGDOMAIN, "BROKER ANNOUNCE_VALUE %s", AQH_Value_GetName(msgValue));
+      AQH_Endpoint_AddMsgOut(xo->brokerEndpoint, pubMsg);
+    }
+    AQH_Value_free(msgValue);
+  }
+  else {
+    DBG_INFO(AQH_LOGDOMAIN, "Ignoring BROKER ANNOUNCE_VALUE %s", AQH_Value_GetName(msgValue));
   }
-  AQH_Value_free(msgValue);
 }
 
 

apps/aqhome-mqttlog/s_setdata.c

@@ -0,0 +1,171 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2025 Martin Preuss, all rights reserved.
+ *
+ * The license for this file can be found in the file COPYING which you
+ * should have received along with this file.
+ ****************************************************************************/
+
+#ifdef HAVE_CONFIG_H
+# include <config.h>
+#endif
+
+#include "./s_setdata.h"
+#include "./server_p.h"
+#include "aqhome/data/value.h"
+#include "aqhome/msg/ipc/data/m_ipcd_setdata.h"
+#include "aqhome/msg/mqtt/m_mqtt_publish.h"
+#include "aqhome/ipc2/endpoint.h"
+
+#include <gwenhywfar/debug.h>
+
+
+
+/* ------------------------------------------------------------------------------------------------
+ * forward declarations
+ * ------------------------------------------------------------------------------------------------
+ */
+
+static void _sendDataForDevice(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *device,
+                               const char *valueName, const char *valueData);
+static void _sendValueToMqtt(AQH_MQTTLOG_SERVER *xo, const char *deviceId,
+                             const AQHMQTT_TOPIC *topic, const char *valueData);
+static GWEN_BUFFER *_createBufferForTopic(const char *deviceId, const AQHMQTT_TOPIC *topic);
+
+
+
+/* ------------------------------------------------------------------------------------------------
+ * implementations
+ * ------------------------------------------------------------------------------------------------
+ */
+
+void AQH_MqttLogServer_HandleSetData(AQH_OBJECT *o,
+                                     const AQH_MESSAGE *msg,
+                                     const GWEN_TAG16_LIST *tagList)
+{
+  if (o && msg) {
+    AQH_MQTTLOG_SERVER *xo;
+
+    xo=AQH_MqttLogServer_GetServerData(o);
+    if (xo) {
+      AQH_VALUE *recvdValue;
+    
+      DBG_ERROR(NULL, "Received SETDATA request");
+      recvdValue=AQH_IpcdMessageSetData_ReadValue(tagList);
+      if (recvdValue) {
+        const char *valueName;
+        const char *deviceName;
+    
+        valueName=recvdValue?AQH_Value_GetName(recvdValue):NULL;
+        deviceName=recvdValue?AQH_Value_GetDeviceName(recvdValue):NULL;
+        if (valueName && deviceName) {
+          AQHMQTT_DEVICE *device;
+    
+          device=AQH_MqttLogServer_FindRegisteredDevice(o, deviceName);
+          if (device) {
+            char *valueDataFreeable;
+    
+            DBG_ERROR(NULL, "Sending data to value \"%s\" of device \"%s\"", valueName, deviceName);
+            valueDataFreeable=AQH_IpcdMessageSetData_ReadData(tagList);
+            _sendDataForDevice(xo, device, valueName, valueDataFreeable);
+            free(valueDataFreeable);
+          }
+          else {
+            DBG_ERROR(NULL, "Device \"%s\" not found", deviceName);
+            AQH_MqttLogServer_DumpRegisteredDevices(o);
+          }
+        }
+        else {
+          DBG_ERROR(NULL, "Either value name or device name missing in request");
+        }
+        AQH_Value_free(recvdValue);
+      }
+      else {
+        DBG_ERROR(NULL, "Request does not contain a value object");
+      }
+    }
+  }
+}
+
+
+
+void _sendDataForDevice(AQH_MQTTLOG_SERVER *xo,
+                        const AQHMQTT_DEVICE *device,
+                        const char *valueName, const char *valueData)
+{
+  const char *deviceId;
+
+  deviceId=AQHMQTT_Device_GetId(device);
+  if (deviceId && *deviceId) {
+    AQHMQTT_TOPIC_LIST *topicList;
+
+    topicList=AQHMQTT_Device_GetTopicList(device);
+    if (topicList) {
+      AQHMQTT_TOPIC *topic;
+
+      topic=AQHMQTT_Topic_List_First(topicList);
+      while(topic) {
+        if (AQHMQTT_Topic_GetDirection(topic)==AQHMQTT_TopicDir_Out) {
+          AQHMQTT_VALUE_LIST *valueList;
+          AQHMQTT_VALUE *value;
+
+          valueList=AQHMQTT_Topic_GetValueList(topic);
+          value=valueList?AQHMQTT_Value_List_GetByName(valueList, valueName):NULL;
+          if (value) {
+            /* found value, create publish msg, send */
+            DBG_ERROR(NULL, "Topic \"%s\" contains value \"%s\"", AQHMQTT_Topic_GetName(topic), valueName);
+            _sendValueToMqtt(xo, deviceId, topic, valueData);
+          }
+        } /* if out */
+        topic=AQHMQTT_Topic_List_Next(topic);
+      } /* while topic */
+    }
+  }
+  else {
+    DBG_ERROR(NULL, "Device has no id");
+  }
+}
+
+
+
+void _sendValueToMqtt(AQH_MQTTLOG_SERVER *xo,  const char *deviceId, const AQHMQTT_TOPIC *topic, const char *valueData)
+{
+  GWEN_BUFFER *buf;
+  AQH_MESSAGE *msgOut;
+
+  buf=_createBufferForTopic(deviceId, topic);
+  DBG_ERROR(NULL, "MQTT PUBLISH: %s = %s", GWEN_Buffer_GetStart(buf), valueData?valueData:"<empty>");
+  msgOut=AQH_MqttMessagePublish_new(0, 0, GWEN_Buffer_GetStart(buf),
+                                    (const uint8_t*) (valueData?valueData:NULL),
+                                    valueData?strlen(valueData):0);
+  if (msgOut)
+    AQH_Endpoint_AddMsgOut(xo->mqttEndpoint, msgOut);
+  else {
+    DBG_ERROR(NULL, "Error creating message");
+  }
+  GWEN_Buffer_free(buf);
+}
+
+
+
+GWEN_BUFFER *_createBufferForTopic(const char *deviceId, const AQHMQTT_TOPIC *topic)
+{
+  GWEN_BUFFER *buf;
+  const char *s;
+
+  buf=GWEN_Buffer_new(0, 256, 0, 1);
+  s=AQHMQTT_Topic_GetBeforeId(topic);
+  if (s && *s)
+    GWEN_Buffer_AppendString(buf, s);
+  GWEN_Buffer_AppendString(buf, deviceId);
+  s=AQHMQTT_Topic_GetAfterId(topic);
+  if (s && *s)
+    GWEN_Buffer_AppendString(buf, s);
+  return buf;
+}
+
+
+
+
+
+

apps/aqhome-mqttlog/s_setdata.h

@@ -0,0 +1,27 @@
+/****************************************************************************
+ * This file is part of the project AqHome.
+ * AqHome (c) by 2025 Martin Preuss, all rights reserved.
+ *
+ * The license for this file can be found in the file COPYING which you
+ * should have received along with this file.
+ ****************************************************************************/
+
+#ifndef AQHOMEMQTT_S_SETDATA_H
+#define AQHOMEMQTT_S_SETDATA_H
+
+
+#include "./aqhome_mqtt.h"
+
+#include <aqhome/events2/object.h>
+#include <aqhome/ipc2/message.h>
+
+#include <gwenhywfar/tag16.h>
+
+
+void AQH_MqttLogServer_HandleSetData(AQH_OBJECT *o, const AQH_MESSAGE *recvdMsg, const GWEN_TAG16_LIST *tagList);
+
+
+
+#endif
+
+

apps/aqhome-mqttlog/server.c

@@ -12,6 +12,7 @@
 
 #include "./server_p.h"
 #include "./s_publish.h"
+#include "./s_setdata.h"
 #include "./xmlread.h"
 #include "./xmlwrite.h"
 
@@ -25,6 +26,7 @@
 #include <aqhome/msg/ipc/m_ipc.h>
 #include <aqhome/msg/ipc/m_ipc_result.h>
 #include <aqhome/msg/ipc/m_ipc_connect.h>
+#include <aqhome/msg/ipc/m_ipc_tag16.h>
 #include <aqhome/msg/ipc/data/m_ipcd.h>
 #include <aqhome/msg/ipc/data/m_ipcd_multidata.h>
 #include <aqhome/ipc2/mqtt_endpoint.h>
@@ -657,21 +659,26 @@ void AQH_MqttLogServer_HandleBrokerMsgs(AQH_OBJECT *o)
 
 void _handleMsgFromBroker(AQH_OBJECT *o, AQH_OBJECT *ep, const AQH_MESSAGE *msg)
 {
-  uint16_t code;
-  uint8_t protoId;
+  GWEN_TAG16_LIST *tagList;
 
-  /* exec IPC message */
-  code=AQH_IpcMessage_GetCode(msg);
-  protoId=AQH_IpcMessage_GetProtoId(msg);
-  if (protoId==AQH_IPC_PROTOCOL_DATA_ID) {
-    DBG_ERROR(NULL, "Received IPC packet %d (%x)", (int) code, code);
-    switch(code) {
-//    case AQH_MSGTYPE_IPC_DATA_SETDATA:  AQH_MqttLogServer_HandleSetData(o, ep, msg); break;
-    default:                            break;
+  tagList=AQH_IpcMessageTag16_ParsePayload(msg, 0);
+  if (tagList) {
+    uint16_t code;
+    uint8_t protoId;
+
+    code=AQH_IpcMessage_GetCode(msg);
+    protoId=AQH_IpcMessage_GetProtoId(msg);
+    if (protoId==AQH_IPC_PROTOCOL_DATA_ID) {
+      DBG_ERROR(NULL, "Received IPC packet %d (%x)", (int) code, code);
+      switch(code) {
+      case AQH_MSGTYPE_IPC_DATA_SETDATA:  AQH_MqttLogServer_HandleSetData(o, msg, tagList); break;
+      default:                            break;
+      }
     }
-  }
-  else {
-    DBG_ERROR(NULL, "Invalid IPC protocol %d (%02x)", protoId, protoId);
+    else {
+      DBG_ERROR(NULL, "Invalid IPC protocol %d (%02x)", protoId, protoId);
+    }
+    GWEN_Tag16_List_free(tagList);
   }
 }