martin/aqhomecontrol
Public Access
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

aqhome-apps: removed unneeded files.

Browse Source
This commit is contained in:
Martin Preuss
2025-03-10 00:15:36 +01:00
parent 5011e7e123
commit dc4a02a8ff
70 changed files with 7 additions and 6824 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
apps/aqhome-mqttlog/0BUILD
View File

@@ -39,7 +39,6 @@
<headers dist="true" >
xmlread.h
xmlwrite.h
c_setdata.h
server.h
server_p.h
s_publish.h
@@ -48,11 +47,9 @@
<sources>
$(local/typefiles)
aqhome_mqtt.c
main.c
xmlread.c
xmlwrite.c
c_setdata.c
server.c
s_publish.c
s_setdata.c

189
apps/aqhome-mqttlog/aqhome_mqtt.c
View File

@@ -1,189 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 "./aqhome_mqtt_p.h"
#include "aqhome/ipc/endpoint_ipc.h"
#include "aqhome/ipc/endpoint_ipcclient.h"
#include "aqhome/mqtt/endpoint_mqttc.h"
#include "aqhome/mqtt/msg_mqtt_publish.h"
#include <gwenhywfar/misc.h>
#include <gwenhywfar/debug.h>
AQHOME_MQTT *AqHomeMqtt_new(void)
{
AQHOME_MQTT *aqh;
GWEN_NEW_OBJECT(AQHOME_MQTT, aqh);
return aqh;
}
void AqHomeMqtt_free(AQHOME_MQTT *aqh)
{
if (aqh) {
AQHMQTT_Device_List_free(aqh->availableDeviceList);
AQHMQTT_Device_List_free(aqh->registeredDeviceList);
GWEN_MsgEndpoint_free(aqh->rootEndpoint);
GWEN_DB_Group_free(aqh->dbArgs);
free(aqh->pidFile);
GWEN_FREE_OBJECT(aqh);
}
}
GWEN_MSG_ENDPOINT *AqHomeMqtt_GetBrokerEndpoint(const AQHOME_MQTT *aqh)
{
return aqh?(aqh->brokerEndpoint):NULL;
}
GWEN_MSG_ENDPOINT *AqHomeMqtt_GetMqttEndpoint(const AQHOME_MQTT *aqh)
{
return aqh?(aqh->mqttEndpoint):NULL;
}
GWEN_DB_NODE *AqHomeMqtt_GetDbArgs(const AQHOME_MQTT *aqh)
{
return aqh?(aqh->dbArgs):NULL;
}
const char *AqHomeMqtt_GetPidFile(const AQHOME_MQTT *aqh)
{
return aqh?aqh->pidFile:NULL;
}
void AqHomeMqtt_SetPidFile(AQHOME_MQTT *aqh, const char *s)
{
if (aqh) {
free(aqh->pidFile);
aqh->pidFile=s?strdup(s):NULL;
}
}
int AqHomeMqtt_GetTimeout(const AQHOME_MQTT *aqh)
{
return aqh?aqh->timeout:0;
}
const char *AqHomeMqtt_GetDeviceFile(const AQHOME_MQTT *aqh)
{
return aqh?aqh->deviceFile:NULL;
}
void AqHomeMqtt_SetDeviceFile(AQHOME_MQTT *aqh, const char *s)
{
if (aqh) {
free(aqh->deviceFile);
aqh->deviceFile=s?strdup(s):NULL;
}
}
AQHMQTT_DEVICE_LIST *AqHomeMqtt_GetAvailableDeviceList(const AQHOME_MQTT *aqh)
{
return aqh?aqh->availableDeviceList:NULL;
}
void AqHomeMqtt_SetAvailableDeviceList(AQHOME_MQTT *aqh, AQHMQTT_DEVICE_LIST *dl)
{
if (aqh) {
AQHMQTT_Device_List_free(aqh->availableDeviceList);
aqh->availableDeviceList=dl;
}
}
void AqHomeMqtt_SetRegisteredDeviceList(AQHOME_MQTT *aqh, AQHMQTT_DEVICE_LIST *dl)
{
if (aqh) {
AQHMQTT_Device_List_free(aqh->registeredDeviceList);
aqh->registeredDeviceList=dl;
}
}
AQHMQTT_DEVICE *AqHomeMqtt_FindRegisteredDevice(AQHOME_MQTT *aqh, const char *wantedDeviceId)
{
if (aqh && aqh->registeredDeviceList) {
return AQHMQTT_Device_List_GetById(aqh->registeredDeviceList, wantedDeviceId);
}
else {
DBG_ERROR(NULL, "No registered devices");
}
return NULL;
}
void AqHomeMqtt_DumpRegisteredDevices(const AQHOME_MQTT *aqh)
{
if (aqh && aqh->registeredDeviceList) {
AQHMQTT_DEVICE *device;
device=AQHMQTT_Device_List_First(aqh->registeredDeviceList);
if (device) {
fprintf(stderr, "Registered Devices:\n");
while(device) {
const char *sDeviceName;
const char *sDeviceId;
sDeviceName=AQHMQTT_Device_GetName(device);
sDeviceId=AQHMQTT_Device_GetId(device);
fprintf(stderr, " %s (%s)\n", sDeviceId?sDeviceId:"<no id>", sDeviceName?sDeviceName:"<no name>");
device=AQHMQTT_Device_List_Next(device);
}
}
else {
fprintf(stderr, "No registered devices\n");
}
}
else {
fprintf(stderr, "No registered devices\n");
}
}

51
apps/aqhome-mqttlog/aqhome_mqtt.h
View File

@@ -1,51 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 AQHOME_MQTT_H
#define AQHOME_MQTT_H
//#include "./mqttvalue.h"
//#include "./mqtttopic.h"
#include "aqhome-mqttlog/types/device.h"
#include <gwenhywfar/endpoint.h>
typedef struct AQHOME_MQTT AQHOME_MQTT;
AQHOME_MQTT *AqHomeMqtt_new(void);
void AqHomeMqtt_free(AQHOME_MQTT *aqh);
GWEN_MSG_ENDPOINT *AqHomeMqtt_GetBrokerEndpoint(const AQHOME_MQTT *aqh);
GWEN_MSG_ENDPOINT *AqHomeMqtt_GetMqttEndpoint(const AQHOME_MQTT *aqh);
GWEN_DB_NODE *AqHomeMqtt_GetDbArgs(const AQHOME_MQTT *aqh);
const char *AqHomeMqtt_GetPidFile(const AQHOME_MQTT *aqh);
void AqHomeMqtt_SetPidFile(AQHOME_MQTT *aqh, const char *s);
int AqHomeMqtt_GetTimeout(const AQHOME_MQTT *aqh);
const char *AqHomeMqtt_GetDeviceFile(const AQHOME_MQTT *aqh);
void AqHomeMqtt_SetDeviceFile(AQHOME_MQTT *aqh, const char *s);
AQHMQTT_DEVICE_LIST *AqHomeMqtt_GetAvailableDeviceList(const AQHOME_MQTT *aqh);
void AqHomeMqtt_SetAvailableDeviceList(AQHOME_MQTT *aqh, AQHMQTT_DEVICE_LIST *dl);
void AqHomeMqtt_SetRegisteredDeviceList(AQHOME_MQTT *aqh, AQHMQTT_DEVICE_LIST *dl);
AQHMQTT_DEVICE *AqHomeMqtt_FindRegisteredDevice(AQHOME_MQTT *aqh, const char *wantedDeviceId);
void AqHomeMqtt_DumpRegisteredDevices(const AQHOME_MQTT *aqh);
#endif

45
apps/aqhome-mqttlog/aqhome_mqtt_p.h
View File

@@ -1,45 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 AQHOME_MQTT_P_H
#define AQHOME_MQTT_P_H
#include "./aqhome_mqtt.h"
#include <gwenhywfar/mutex.h>
#define AQHOME_MQTT_DEFAULT_PIDFILE "/var/run/aqhome-mqtt.pid"
#define AQHOME_MQTT_DEFAULT_DATADIR "/var/lib/aqhome-mqtt"
#define AQHOME_MQTT_DEFAULT_DEVICEFILE "mqttlog/registereddevices.xml"
#define AQHOME_MQTT_DEFAULT_BROKER_PORT 1899
#define AQHOME_MQTT_DEFAULT_BROKER_CLIENTID "mqtt"
struct AQHOME_MQTT {
GWEN_MSG_ENDPOINT *rootEndpoint;
GWEN_MSG_ENDPOINT *brokerEndpoint; /* do not free (is part of tree pointed to by rootEndpoint) */
GWEN_MSG_ENDPOINT *mqttEndpoint; /* do not free (is part of tree pointed to by rootEndpoint) */
GWEN_DB_NODE *dbArgs;
char *pidFile;
int timeout; /* timeout for run e.g. inside valgrind */
AQHMQTT_DEVICE_LIST *availableDeviceList;
AQHMQTT_DEVICE_LIST *registeredDeviceList;
char *deviceFile;
};
#endif

160
apps/aqhome-mqttlog/c_setdata.c
View File

@@ -1,160 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2024 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 "./c_setdata.h"
#include "aqhome/data/value.h"
#include "aqhome/ipc/data/msg_data_set.h"
#include "aqhome/mqtt/msg_mqtt_publish.h"
#include <gwenhywfar/debug.h>
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static void _sendDataForDevice(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device, const char *valueName, const char *valueData);
static void _sendValueToMqtt(AQHOME_MQTT *aqh, const char *deviceId, const AQHMQTT_TOPIC *topic, const char *valueData);
static GWEN_BUFFER *_createBufferForTopic(const char *deviceId, const AQHMQTT_TOPIC *topic);
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
void AqHomeMqttLog_HandleSetData(AQHOME_MQTT *aqh, GWEN_UNUSED GWEN_MSG_ENDPOINT *ep, GWEN_MSG *recvdMsg)
{
AQH_VALUE *recvdValue;
DBG_ERROR(NULL, "Received SETDATA request");
AQH_SetDataIpcMsg_Parse(recvdMsg, 0);
recvdValue=AQH_SetDataIpcMsg_ReadValue(recvdMsg);
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=AqHomeMqtt_FindRegisteredDevice(aqh, deviceName);
if (device) {
char *valueDataFreeable;
DBG_ERROR(NULL, "Sending data to value \"%s\" of device \"%s\"", valueName, deviceName);
valueDataFreeable=AQH_SetDataIpcMsg_ReadData(recvdMsg);
_sendDataForDevice(aqh, device, valueName, valueDataFreeable);
free(valueDataFreeable);
}
else {
DBG_ERROR(NULL, "Device \"%s\" not found", deviceName);
AqHomeMqtt_DumpRegisteredDevices(aqh);
}
}
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(AQHOME_MQTT *aqh, 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(aqh, deviceId, topic, valueData);
}
} /* if out */
topic=AQHMQTT_Topic_List_Next(topic);
} /* while topic */
}
}
else {
DBG_ERROR(NULL, "Device has no id");
}
}
void _sendValueToMqtt(AQHOME_MQTT *aqh, const char *deviceId, const AQHMQTT_TOPIC *topic, const char *valueData)
{
GWEN_MSG_ENDPOINT *ep;
GWEN_BUFFER *buf;
GWEN_MSG *msgOut;
ep=AqHomeMqtt_GetMqttEndpoint(aqh);
buf=_createBufferForTopic(deviceId, topic);
DBG_ERROR(NULL, "MQTT PUBLISH: %s = %s", GWEN_Buffer_GetStart(buf), valueData?valueData:"<empty>");
msgOut=AQH_PublishMqttMsg_new(0, 0, GWEN_Buffer_GetStart(buf),
(const uint8_t*) (valueData?valueData:NULL),
valueData?strlen(valueData):0);
if (msgOut) {
GWEN_MsgEndpoint_AddSendMessage(ep, 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;
}

25
apps/aqhome-mqttlog/c_setdata.h
View File

@@ -1,25 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 AQHOME_MQTTLOG_C_SETDATA_H
#define AQHOME_MQTTLOG_C_SETDATA_H
#include "./aqhome_mqtt.h"
void AqHomeMqttLog_HandleSetData(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, GWEN_MSG *recvdMsg);
#endif

85
apps/aqhome-mqttlog/fini.c
View File

@@ -1,85 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 "./fini.h"
#include "./aqhome_mqtt_p.h"
#include <gwenhywfar/gwenhywfar.h>
#include <gwenhywfar/args.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/endpoint.h>
#include <unistd.h>
/* ------------------------------------------------------------------------------------------------
* defines
* ------------------------------------------------------------------------------------------------
*/
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static void _disconnectTree(GWEN_MSG_ENDPOINT *ep);
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
void AqHomeMqtt_Fini(AQHOME_MQTT *aqh)
{
if (aqh) {
if (aqh->rootEndpoint)
_disconnectTree(aqh->rootEndpoint);
GWEN_MsgEndpoint_free(aqh->rootEndpoint);
aqh->rootEndpoint=NULL;
aqh->brokerEndpoint=NULL;
aqh->mqttEndpoint=NULL;
AQHMQTT_Device_List_free(aqh->availableDeviceList);
aqh->availableDeviceList=NULL;
AQHMQTT_Device_List_free(aqh->registeredDeviceList);
aqh->registeredDeviceList=NULL;
if (aqh->pidFile)
remove(aqh->pidFile);
}
}
void _disconnectTree(GWEN_MSG_ENDPOINT *ep)
{
GWEN_MSG_ENDPOINT *epChild;
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
while(epChild) {
_disconnectTree(epChild);
epChild=GWEN_MsgEndpoint_Tree2_GetNext(epChild);
} /* while */
GWEN_MsgEndpoint_Disconnect(ep);
}

23
apps/aqhome-mqttlog/fini.h
View File

@@ -1,23 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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_FINI_H
#define AQHOMEMQTT_FINI_H
#include "./aqhome_mqtt.h"
void AqHomeMqtt_Fini(AQHOME_MQTT *aqh);
#endif

489
apps/aqhome-mqttlog/init.c
View File

@@ -1,489 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 "./init.h"
#include "./aqhome_mqtt_p.h"
#include "./xmlread.h"
#include "./xmlwrite.h"
#include "aqhome/aqhome.h"
#include "aqhome/ipc/endpoint_ipc.h"
#include "aqhome/ipc/endpoint_ipcclient.h"
#include "aqhome/mqtt/endpoint_mqttc.h"
#include <gwenhywfar/endpoint_multilayer.h>
#include <gwenhywfar/args.h>
#include <gwenhywfar/debug.h>
#include <unistd.h>
#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#include <string.h>
#include <time.h>
/* ------------------------------------------------------------------------------------------------
* defines
* ------------------------------------------------------------------------------------------------
*/
#define I18N(msg) msg
#define I18S(msg) msg
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static int _createPidFile(const char *pidFilename);
static int _setupBroker(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs);
static int _setupMqtt(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs);
static int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs);
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
int AqHomeMqtt_Init(AQHOME_MQTT *aqh, int argc, char **argv)
{
int rv;
GWEN_DB_NODE *dbArgs;
const char *s;
dbArgs=GWEN_DB_Group_new("args");
rv=_readArgs(argc, argv, dbArgs);
if (rv<0) {
DBG_ERROR(NULL, "Error reading args (%d)", rv);
return rv;
}
AQH_MergeConfigFileIntoConfig(dbArgs, "ConfigFile");
aqh->dbArgs=dbArgs;
s=GWEN_DB_GetCharValue(dbArgs, "loglevel", 0, NULL);
if (s && *s) {
GWEN_LOGGER_LEVEL ll;
ll=GWEN_Logger_Name2Level(s);
GWEN_Logger_SetLevel(NULL, ll);
}
aqh->timeout=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 0);
s=GWEN_DB_GetCharValue(dbArgs, "pidfile", 0, AQHOME_MQTT_DEFAULT_PIDFILE);
if (s && *s) {
AqHomeMqtt_SetPidFile(aqh, s);
rv=_createPidFile(s);
if (rv<0) {
DBG_ERROR(NULL, "Error creating PID file (%d)", rv);
return rv;
}
}
s=GWEN_DB_GetCharValue(dbArgs, "devicefile", 0, NULL);
if (s && *s) {
AqHomeMqtt_SetDeviceFile(aqh, s);
}
else {
GWEN_BUFFER *bufFilename;
bufFilename=AQH_GetRuntimeFilePath(AQHOME_MQTT_DEFAULT_DEVICEFILE);
if (bufFilename) {
AqHomeMqtt_SetDeviceFile(aqh, GWEN_Buffer_GetStart(bufFilename));
GWEN_Buffer_free(bufFilename);
}
else {
DBG_ERROR(NULL, "Could not setup filename for devices, please specify via command line argument");
return GWEN_ERROR_GENERIC;
}
}
aqh->rootEndpoint=GWEN_MsgEndpoint_new("root", 0);
rv=_setupMqtt(aqh, dbArgs);
if (rv<0) {
DBG_ERROR(NULL, "Error setting up connection to broker (%d)", rv);
return rv;
}
rv=_setupBroker(aqh, dbArgs);
if (rv<0) {
DBG_ERROR(NULL, "Error setting up connection to broker (%d)", rv);
return rv;
}
AqHomeMqtt_LoadRuntimeDeviceFiles(aqh);
AqHomeMqtt_ReloadDeviceFiles(aqh);
return 0;
}
void AqHomeMqtt_ReloadDeviceFiles(AQHOME_MQTT *aqh)
{
AQHMQTT_DEVICE_LIST *deviceList;
DBG_ERROR(NULL, "Loading devices description files");
deviceList=AqHomeMqttLog_ReadDataDeviceFiles(aqh);
if (deviceList)
AqHomeMqtt_SetAvailableDeviceList(aqh, deviceList);
}
void AqHomeMqtt_LoadRuntimeDeviceFiles(AQHOME_MQTT *aqh)
{
AQHMQTT_DEVICE_LIST *deviceList;
DBG_ERROR(NULL, "Loading registered devices from file \"%s\"", aqh->deviceFile);
deviceList=AqHomeMqttLog_ReadDeviceFile(aqh, aqh->deviceFile);
if (deviceList)
AqHomeMqtt_SetRegisteredDeviceList(aqh, deviceList);
}
int AqHomeMqtt_SaveRuntimeDeviceFiles(AQHOME_MQTT *aqh)
{
int rv;
rv=AqHomeMqttLog_WriteDevicesFile(aqh, aqh->registeredDeviceList, aqh->deviceFile);
if (rv<0) {
DBG_INFO(NULL, "Error writing devices to \"%s\" (%d)", aqh->deviceFile, rv);
return rv;
}
return 0;
}
int _createPidFile(const char *pidFilename)
{
FILE *f;
int pidfd;
if (remove(pidFilename)==0) {
DBG_ERROR(0, "Old PID file existed, removed. (Unclean shutdown?)");
}
#ifdef HAVE_SYS_STAT_H
pidfd = open(pidFilename, O_EXCL|O_CREAT|O_WRONLY, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
if (pidfd < 0) {
DBG_ERROR(NULL, "Could not create PID file \"%s\" (%s), aborting.", pidFilename, strerror(errno));
return GWEN_ERROR_IO;
}
f = fdopen(pidfd, "w");
#else /* HAVE_STAT_H */
f=fopen(pidFilename,"w+");
#endif /* HAVE_STAT_H */
/* write pid */
#ifdef HAVE_GETPID
fprintf(f,"%d\n",getpid());
#else
fprintf(f,"-1\n");
#endif
if (fclose(f)) {
DBG_ERROR(0, "Could not close PID file \"%s\" (%s), aborting.", pidFilename, strerror(errno));
return GWEN_ERROR_IO;
}
return 0;
}
int _setupBroker(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs)
{
const char *brokerAddress;
int brokerPort;
const char *brokerClientId;
brokerAddress=GWEN_DB_GetCharValue(dbArgs, "brokerAddress", 0, NULL);
if (!(brokerAddress && *brokerAddress))
brokerAddress=GWEN_DB_GetCharValue(dbArgs, "ConfigFile/brokerAddress", 0, "127.0.0.1");
brokerPort=GWEN_DB_GetIntValue(dbArgs, "brokerPort", 0, -1);
if (brokerPort<0)
brokerPort=GWEN_DB_GetIntValue(dbArgs, "ConfigFile/brokerPort", 0, AQHOME_MQTT_DEFAULT_BROKER_PORT);
brokerClientId=GWEN_DB_GetCharValue(dbArgs, "brokerClientId", 0, AQHOME_MQTT_DEFAULT_BROKER_CLIENTID);
if (brokerAddress && *brokerAddress && brokerPort) {
GWEN_MSG_ENDPOINT *ep;
GWEN_MSG_ENDPOINT *ipcBaseEndpoint;
int rv;
ep=AQH_ClientIpcEndpoint_new("brokerIpcClient", 0);
ipcBaseEndpoint=AQH_IpcEndpoint_CreateIpcTcpClient(brokerAddress, brokerPort, "brokerPhysEndpoint", 0);
AQH_IpcEndpoint_SetServiceName(ipcBaseEndpoint, brokerClientId);
GWEN_MsgEndpoint_Tree2_AddChild(ep, ipcBaseEndpoint);
GWEN_MsgEndpoint_Tree2_AddChild(aqh->rootEndpoint, ep);
aqh->brokerEndpoint=ep;
rv=GWEN_MultilayerEndpoint_StartConnect(ep);
if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) {
DBG_ERROR(NULL, "Error connecting to broker server %s:%d (%d), will retry later", brokerAddress, brokerPort, rv);
return rv;
}
}
return 0;
}
int _setupMqtt(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs)
{
const char *mqttAddress;
int mqttPort;
const char *mqttClientId;
int mqttKeepAlive;
mqttAddress=GWEN_DB_GetCharValue(dbArgs, "mqttAddress", 0, NULL);
if (!(mqttAddress && *mqttAddress))
mqttAddress=GWEN_DB_GetCharValue(dbArgs, "ConfigFile/mqttAddr", 0, "127.0.0.1");
mqttPort=GWEN_DB_GetIntValue(dbArgs, "mqttPort", 0, 1883);
if (mqttPort<0)
mqttPort=GWEN_DB_GetIntValue(dbArgs, "ConfigFile/mqttPort", 0, 1883);
mqttClientId=GWEN_DB_GetCharValue(dbArgs, "mqttClientId", 0, "aqhome-mqttlog");
if (!(mqttClientId && *mqttClientId))
mqttClientId=GWEN_DB_GetCharValue(dbArgs, "ConfigFile/mqttClientId", 0, "aqhome-mqttlog");
mqttKeepAlive=GWEN_DB_GetIntValue(dbArgs, "mqttKeepAlive", 0, 600);
if (mqttAddress && *mqttAddress && mqttPort) {
GWEN_MSG_ENDPOINT *epMqtt;
DBG_INFO(AQH_LOGDOMAIN, "Connecting to %s (port %d)", mqttAddress, mqttPort);
epMqtt=AQH_MqttClientEndpoint_new(mqttClientId, mqttAddress, mqttPort, NULL, 0);
if (epMqtt==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint TCP");
return GWEN_ERROR_IO;
}
AQH_MqttClientEndpoint_SetKeepAliveTime(epMqtt, mqttKeepAlive);
GWEN_MsgEndpoint_AddFlags(epMqtt, AQH_ENDPOINT2_MQTTCLIENT_FLAGS_SUBSCRIBEALL);
GWEN_MsgEndpoint_Tree2_AddChild(aqh->rootEndpoint, epMqtt);
aqh->mqttEndpoint=epMqtt;
}
return 0;
}
int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs)
{
int rv;
const GWEN_ARGS args[]= {
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"loglevel", /* name */
0, /* minnum */
1, /* maxnum */
"L", /* short option */
"loglevel", /* long option */
I18S("Specify loglevel"), /* short description */
I18S("Specify loglevel") /* long description */
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"cfgdir", /* name */
0, /* minnum */
1, /* maxnum */
"D", /* short option */
"cfgdir", /* long option */
I18S("Specify the configuration folder"),
I18S("Specify the configuration folder")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"charset", /* name */
0, /* minnum */
1, /* maxnum */
0, /* short option */
"charset", /* long option */
I18S("Specify the output character set"), /* short description */
I18S("Specify the output character set") /* long description */
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"brokerAddress", /* name */
0, /* minnum */
1, /* maxnum */
"ba", /* short option */
"brokeraddress", /* long option */
I18S("Specify the address of the broker server to connect to (disabled if missing)"),
I18S("Specify the address of the broker server to connect to (disabled if missing)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"brokerPort", /* name */
0, /* minnum */
1, /* maxnum */
"bp", /* short option */
"brokerport", /* long option */
I18S("Specify the port of the broker server (default: 1899)"),
I18S("Specify the port of the broker server (default: 1899)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"brokerClientId", /* name */
0, /* minnum */
1, /* maxnum */
NULL, /* short option */
"brokerclientid", /* long option */
I18S("Specify client id for the broker server (default: \"nodes\")"),
I18S("Specify client id for the broker server (default: \"nodes\")")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"mqttAddress", /* name */
0, /* minnum */
1, /* maxnum */
"ma", /* short option */
"mqttaddress", /* long option */
I18S("Specify the address of the MQTT server to connect to (disabled if missing)"),
I18S("Specify the address of the MQTT server to connect to (disabled if missing)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"mqttPort", /* name */
0, /* minnum */
1, /* maxnum */
"mp", /* short option */
"mqttport", /* long option */
I18S("Specify the port of the MQTT server (default: 1883)"),
I18S("Specify the port of the MQTT server (default: 1883)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"mqttClientId", /* name */
0, /* minnum */
1, /* maxnum */
NULL, /* short option */
"mqttclientid", /* long option */
I18S("Specify client id for the MQTT server (default: \"aqhomed\")"),
I18S("Specify client id for the MQTT server (default: \"aqhomed\")")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"mqttKeepAlive", /* name */
0, /* minnum */
1, /* maxnum */
"mk", /* short option */
"mqttkeepalive", /* long option */
I18S("Specify keepalive time in seconds (defaults: 600)"),
I18S("Specify keepalive time in seconds (defaults: 600)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"pidfile", /* name */
0, /* minnum */
1, /* maxnum */
"p", /* short option */
"pidfile", /* long option */
I18S("Specify the PID file"),
I18S("Specify the PID file")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"timeout", /* name */
0, /* minnum */
1, /* maxnum */
"T", /* short option */
"timeout", /* long option */
I18S("Specify timeout in second (default: no timeout)"),
I18S("Specify timeout in second (default: no timeout)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"devicefile", /* name */
0, /* minnum */
1, /* maxnum */
"d", /* short option */
"devicefile", /* long option */
I18S("Specify the device file"),
I18S("Specify the device file")
},
{
GWEN_ARGS_FLAGS_HELP | GWEN_ARGS_FLAGS_LAST, /* flags */
GWEN_ArgsType_Int, /* type */
"help", /* name */
0, /* minnum */
0, /* maxnum */
"h", /* short option */
"help",
I18S("Show this help screen."),
I18S("Show this help screen.")
}
};
rv=GWEN_Args_Check(argc, argv, 1, 0, args, dbArgs);
if (rv==GWEN_ARGS_RESULT_ERROR) {
fprintf(stderr, "ERROR: Could not parse arguments main\n");
return GWEN_ERROR_INVALID;
}
else if (rv==GWEN_ARGS_RESULT_HELP) {
GWEN_BUFFER *ubuf;
ubuf=GWEN_Buffer_new(0, 1024, 0, 1);
GWEN_Buffer_AppendArgs(ubuf,
I18N("This is version %s.\nUsage: %s [OPTIONS]\n\nOptions:\n"),
AQHOME_VERSION_STRING,
argv[0]);
if (GWEN_Args_Usage(args, ubuf, GWEN_ArgsOutType_Txt)) {
fprintf(stderr, "ERROR: Could not create help string\n");
return 1;
}
GWEN_Buffer_AppendString(ubuf, "\n");
fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(ubuf));
GWEN_Buffer_free(ubuf);
return GWEN_ERROR_CLOSE;
}
return 0;
}

27
apps/aqhome-mqttlog/init.h
View File

@@ -1,27 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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_INIT_H
#define AQHOMEMQTT_INIT_H
#include "./aqhome_mqtt.h"
int AqHomeMqtt_Init(AQHOME_MQTT *aqh, int argc, char **argv);
void AqHomeMqtt_ReloadDeviceFiles(AQHOME_MQTT *aqh);
void AqHomeMqtt_LoadRuntimeDeviceFiles(AQHOME_MQTT *aqh);
int AqHomeMqtt_SaveRuntimeDeviceFiles(AQHOME_MQTT *aqh);
#endif

57
apps/aqhome-mqttlog/loop.c
View File

@@ -1,57 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 "./loop.h"
#include "./loop_ipc.h"
#include "./loop_mqtt.h"
#include "./c_setdata.h"
#include "./aqhome_mqtt_p.h"
#include <gwenhywfar/gwenhywfar.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/text.h>
#include <gwenhywfar/json_read.h>
#include <gwenhywfar/db.h>
//#define FULL_DEBUG
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
void AqHomeMqttLog_Loop(AQHOME_MQTT *aqh, int timeoutInMsecs)
{
if (aqh) {
GWEN_MsgEndpoint_ChildrenIoLoop(aqh->rootEndpoint, timeoutInMsecs);
AqHomeMqttLog_ReadAndHandleMqttMessages(aqh);
AqHomeMqttLog_ReadAndHandleIpcMessages(aqh);
}
}

26
apps/aqhome-mqttlog/loop.h
View File

@@ -1,26 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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_LOOP_H
#define AQHOMEMQTT_LOOP_H
#include "./aqhome_mqtt.h"
#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/msg.h>
void AqHomeMqttLog_Loop(AQHOME_MQTT *aqh, int timeoutInMsecs);
#endif

87
apps/aqhome-mqttlog/loop_ipc.c
View File

@@ -1,87 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 "./loop_ipc.h"
#include "./aqhome_mqtt_p.h"
#include "./c_setdata.h"
#include "aqhome/ipc/data/ipc_data.h"
#include <gwenhywfar/gwenhywfar.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/text.h>
#include <gwenhywfar/json_read.h>
#include <gwenhywfar/db.h>
#define FULL_DEBUG
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static void _handleIpcMsg(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, GWEN_MSG *msg);
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
void AqHomeMqttLog_ReadAndHandleIpcMessages(AQHOME_MQTT *aqh)
{
GWEN_MSG_ENDPOINT *epTcp;
GWEN_MSG *msg;
epTcp=aqh->brokerEndpoint;
while( (msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp)) ) {
_handleIpcMsg(aqh, epTcp, msg);
GWEN_Msg_free(msg);
}
}
void _handleIpcMsg(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, GWEN_MSG *msg)
{
uint16_t code;
uint8_t protoId;
/* exec IPC message */
code=GWEN_IpcMsg_GetCode(msg);
protoId=GWEN_IpcMsg_GetProtoId(msg);
if (protoId==AQH_IPC_PROTOCOL_DATA_ID) {
DBG_DEBUG(AQH_LOGDOMAIN, "Received IPC packet %d (%x)", (int) code, code);
switch(code) {
case AQH_MSGTYPE_IPC_DATA_SETDATA: AqHomeMqttLog_HandleSetData(aqh, ep, msg); break;
default: break;
}
}
else if (protoId==0 && code==AQH_MSGTYPE_IPC_DATA_RESULT) {
/* result received */
}
else {
DBG_ERROR(NULL, "Invalid IPC protocol %d (%02x)", protoId, protoId);
}
}

26
apps/aqhome-mqttlog/loop_ipc.h
View File

@@ -1,26 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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_LOOP_IPC_H
#define AQHOMEMQTT_LOOP_IPC_H
#include "./aqhome_mqtt.h"
#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/msg.h>
void AqHomeMqttLog_ReadAndHandleIpcMessages(AQHOME_MQTT *aqh);
#endif

501
apps/aqhome-mqttlog/loop_mqtt.c
View File

@@ -1,501 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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 "./loop_mqtt.h"
#include "./aqhome_mqtt_p.h"
#include "aqhome/aqhome.h"
#include "aqhome/mqtt/msg_mqtt_publish.h"
#include "aqhome/ipc/data/msg_data_multidata.h"
#include "aqhome/ipc/data/msg_data_values.h"
#include "aqhome/ipc/data/ipc_data.h"
#include <gwenhywfar/gwenhywfar.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/text.h>
#include <gwenhywfar/json_read.h>
#include <gwenhywfar/db.h>
//#define FULL_DEBUG
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static void _handleMqttMsg(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _handlePublishMsg(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static int _handlePublish(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const char *topic, const char *value);
static void _handleNumTopic(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, AQHMQTT_DEVICE *dev, AQHMQTT_TOPIC *t, const char *rcvdValue);
static void _handleJsonTopic(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, AQHMQTT_DEVICE *dev, AQHMQTT_TOPIC *t, const char *rcvdValue);
static void _sendMessage(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value, const char *rcvdValue);
static void _announceDeviceToBroker(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device);
static void _sendAnnounceValueMessage(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value);
static AQH_VALUE *_mkMessageValue(const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value);
static int _mqttValueTypeMessageValueType(int t);
static int _registerNewDeviceForTopic(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const char *rcvdTopic, const char *rcvdValue);
static AQHMQTT_TOPIC *_findMaskMatchingTopic(AQHMQTT_TOPIC_LIST *topicList, const char *rcvdTopic, int dir);
static AQHMQTT_TOPIC *_findTopicMatchingTopic(AQHMQTT_TOPIC_LIST *topicList, const char *rcvdTopic, int dir);
static GWEN_BUFFER *_extractDeviceId(const AQHMQTT_TOPIC *topic, const char *rcvdTopic);
/* ------------------------------------------------------------------------------------------------
* implementations
* ------------------------------------------------------------------------------------------------
*/
void AqHomeMqttLog_ReadAndHandleMqttMessages(AQHOME_MQTT *aqh)
{
GWEN_MSG_ENDPOINT *epTcp;
GWEN_MSG *msg;
epTcp=aqh->mqttEndpoint;
while( (msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp)) ) {
#ifdef FULL_DEBUG
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
#endif
_handleMqttMsg(aqh, epTcp, msg);
GWEN_Msg_free(msg);
}
}
int AqHomeMqttLog_SendPing(AQHOME_MQTT *aqh)
{
GWEN_MSG_ENDPOINT *epTcp;
GWEN_MSG *msgOut;
DBG_INFO(AQH_LOGDOMAIN, "Sending PING");
epTcp=aqh->mqttEndpoint;
msgOut=GWEN_MqttMsg_new(AQH_MQTTMSG_MSGTYPE_PINGREQ, 0, NULL);
if (msgOut==NULL) {
DBG_ERROR(NULL, "Error creating message");
return GWEN_ERROR_INTERNAL;
}
GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut);
return 0;
}
void _handleMqttMsg(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
{
if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(AQH_MQTTMSG_MSGTYPE_PUBLISH & 0xf0)) {
DBG_INFO(AQH_LOGDOMAIN, "PUBLISH message received");
#ifdef FULL_DEBUG
GWEN_BUFFER *buf;
buf=GWEN_Buffer_new(0, 256, 0, 1);
AQH_PublishMqttMsg_DumpToBuffer(msg, buf, "received");
fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(buf));
GWEN_Buffer_free(buf);
#endif
_handlePublishMsg(aqh, ep, msg);
}
else if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(AQH_MQTTMSG_MSGTYPE_PINGRESP & 0xf0)) {
DBG_INFO(AQH_LOGDOMAIN, "PING response received");
}
else {
#ifdef FULL_DEBUG
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
#endif
}
}
void _handlePublishMsg(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
{
char *topic;
char *value;
topic=AQH_PublishMqttMsg_ExtractTopic(msg);
value=AQH_PublishMqttMsg_ExtractValue(msg);
if (topic && value) {
int rv;
rv=_handlePublish(aqh, ep, topic, value);
if (rv!=1) {
DBG_INFO(NULL, "New topic \"%s\", trying to register", topic);
rv=_registerNewDeviceForTopic(aqh, ep, topic, value);
if (rv==1) {
rv=_handlePublish(aqh, ep, topic, value);
if (rv!=1) {
DBG_ERROR(NULL, "Topic \"%s\" still not handled, SNH!", topic);
}
}
}
}
else {
DBG_ERROR(NULL, "Either topic or value missing in PUBLISH msg");
}
free(value);
free(topic);
}
int _handlePublish(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const char *rcvdTopic, const char *rcvdValue)
{
if (rcvdTopic && *rcvdTopic) {
if (aqh->registeredDeviceList) {
AQHMQTT_DEVICE *device;
device=AQHMQTT_Device_List_First(aqh->registeredDeviceList);
while(device) {
AQHMQTT_TOPIC_LIST *topicList;
const char *sDeviceName;
const char *sDeviceId;
sDeviceName=AQHMQTT_Device_GetName(device);
sDeviceId=AQHMQTT_Device_GetId(device);
topicList=AQHMQTT_Device_GetTopicList(device);
if (topicList) {
AQHMQTT_TOPIC *topic;
topic=_findTopicMatchingTopic(topicList, rcvdTopic, AQHMQTT_TopicDir_In);
#if 0
if (topic==NULL) {
topic=_findMaskMatchingTopic(topicList, rcvdTopic, AQHMQTT_TopicDir_In);
if (topic)
AQHMQTT_Topic_SetTopic(topic, rcvdTopic);
}
#endif
if (topic) {
DBG_INFO(AQH_LOGDOMAIN,
"Handling topic \"%s\" for device type %s (id: %s)",
rcvdTopic,
sDeviceName, sDeviceId?sDeviceId:"<no id>");
if (AQHMQTT_Topic_GetTopicType(topic)==AQHMQTT_TopicType_Json)
_handleJsonTopic(aqh, ep, device, topic, rcvdValue);
else
_handleNumTopic(aqh, ep, device, topic, rcvdValue);
return 1;
}
}
device=AQHMQTT_Device_List_Next(device);
}
}
DBG_INFO(AQH_LOGDOMAIN, "ignoring topic \"%s\"", rcvdTopic);
}
return 0;
}
void _handleNumTopic(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, AQHMQTT_DEVICE *device, AQHMQTT_TOPIC *topic, const char *rcvdValue)
{
AQHMQTT_VALUE_LIST *valueList;
valueList=AQHMQTT_Topic_GetValueList(topic);
if (valueList)
_sendMessage(aqh, device, AQHMQTT_Value_List_First(valueList), rcvdValue);
else {
DBG_INFO(NULL, "No value list in device \"%s\"", AQHMQTT_Device_GetId(device));
}
}
void _handleJsonTopic(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, AQHMQTT_DEVICE *device, AQHMQTT_TOPIC *topic, const char *rcvdValue)
{
GWEN_JSON_ELEM *jeRoot;
jeRoot=GWEN_JsonElement_fromString(rcvdValue);
if (jeRoot==NULL) {
DBG_INFO(NULL, "Could not parse JSON value: %s", rcvdValue);
}
else {
AQHMQTT_VALUE_LIST *valueList;
valueList=AQHMQTT_Topic_GetValueList(topic);
if (valueList) {
AQHMQTT_VALUE *value;
value=AQHMQTT_Value_List_First(valueList);
while(value) {
const char *path;
path=AQHMQTT_Value_GetPath(value);
if (path && *path) {
GWEN_JSON_ELEM *je;
je=GWEN_JsonElement_GetElementByPath(jeRoot, path, 0);
if (je) {
const char *s;
s=GWEN_JsonElement_GetData(je);
if (s && *s)
_sendMessage(aqh, device, value, s);
}
}
value=AQHMQTT_Value_List_Next(value);
} /* while */
}
GWEN_JsonElement_free(jeRoot);
}
}
void _sendMessage(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value, const char *rcvdValue)
{
int rv;
union {double f; uint64_t i;} u;
const char *deviceName;
deviceName=AQHMQTT_Device_GetId(device);
rv=GWEN_Text_StringToDouble(rcvdValue, &(u.f));
if (rv<0) {
DBG_ERROR(NULL, "Invalid value received from MQTT server (%s)", rcvdValue?rcvdValue:"<empty>");
}
else {
GWEN_MSG *pubMsg;
uint64_t arrayToSend[2];
AQH_VALUE *msgValue;
arrayToSend[0]=(uint64_t) time(NULL);
arrayToSend[1]=u.i;
msgValue=_mkMessageValue(device, value);
pubMsg=AQH_MultiDataDataIpcMsg_new(AQH_MSGTYPE_IPC_DATA_UPDATEDATA,
GWEN_MsgEndpoint_GetNextMessageId(aqh->brokerEndpoint), 0,
msgValue, arrayToSend, 1);
if (pubMsg) {
DBG_INFO(AQH_LOGDOMAIN, "BROKER UPDATE_DATA %s/%s: %f",
deviceName?deviceName:"<no device name>",
AQH_Value_GetName(msgValue), u.f);
GWEN_MsgEndpoint_AddSendMessage(aqh->brokerEndpoint, pubMsg);
}
AQH_Value_free(msgValue);
}
}
void _announceDeviceToBroker(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device)
{
AQHMQTT_TOPIC_LIST *topicList;
topicList=AQHMQTT_Device_GetTopicList(device);
if (topicList) {
AQHMQTT_TOPIC *topic;
topic=AQHMQTT_Topic_List_First(topicList);
while(topic) {
AQHMQTT_VALUE_LIST *valueList;
valueList=AQHMQTT_Topic_GetValueList(topic);
if (valueList) {
const AQHMQTT_VALUE *value;
value=AQHMQTT_Value_List_First(valueList);
while(value) {
_sendAnnounceValueMessage(aqh, device, value);
value=AQHMQTT_Value_List_Next(value);
}
}
topic=AQHMQTT_Topic_List_Next(topic);
}
}
}
void _sendAnnounceValueMessage(AQHOME_MQTT *aqh, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value)
{
GWEN_MSG *pubMsg;
AQH_VALUE *msgValue;
msgValue=_mkMessageValue(device, value);
pubMsg=AQH_ValuesDataIpcMsg_newForOneValue(AQH_MSGTYPE_IPC_DATA_ANNOUNCEVALUE,
GWEN_MsgEndpoint_GetNextMessageId(aqh->brokerEndpoint), 0,
0, msgValue);
if (pubMsg) {
DBG_INFO(AQH_LOGDOMAIN, "BROKER ANNOUNCE_VALUE %s", AQH_Value_GetName(msgValue));
GWEN_MsgEndpoint_AddSendMessage(aqh->brokerEndpoint, pubMsg);
}
AQH_Value_free(msgValue);
}
AQH_VALUE *_mkMessageValue(const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value)
{
AQH_VALUE *msgValue;
msgValue=AQH_Value_new();
AQH_Value_SetDeviceName(msgValue, AQHMQTT_Device_GetId(device));
AQH_Value_SetName(msgValue, AQHMQTT_Value_GetName(value));
AQH_Value_SetValueUnits(msgValue, AQHMQTT_Value_GetValueUnits(value));
AQH_Value_SetValueType(msgValue, _mqttValueTypeMessageValueType(AQHMQTT_Value_GetValueType(value)));
return msgValue;
}
int _mqttValueTypeMessageValueType(int t)
{
switch(t){
case AQHMQTT_ValueType_Sensor: return AQH_ValueType_Sensor;
case AQHMQTT_ValueType_Actor: return AQH_ValueType_Actor;
default: break;
}
DBG_ERROR(AQH_LOGDOMAIN, "Invalid mqtt value type %d", t);
return AQH_ValueType_Sensor;
}
int _registerNewDeviceForTopic(AQHOME_MQTT *aqh, GWEN_MSG_ENDPOINT *ep, const char *rcvdTopic, const char *rcvdValue)
{
if (rcvdTopic && *rcvdTopic) {
if (aqh->availableDeviceList) {
AQHMQTT_DEVICE *device;
device=AQHMQTT_Device_List_First(aqh->availableDeviceList);
while(device) {
AQHMQTT_TOPIC_LIST *topicList;
topicList=AQHMQTT_Device_GetTopicList(device);
if (topicList) {
AQHMQTT_TOPIC *topic;
topic=_findMaskMatchingTopic(topicList, rcvdTopic, AQHMQTT_TopicDir_In);
if (topic) {
GWEN_BUFFER *buf;
buf=_extractDeviceId(topic, rcvdTopic);
if (buf) {
AQHMQTT_DEVICE *newDevice;
newDevice=AQHMQTT_Device_dup(device);
AQHMQTT_Device_SetId(newDevice, GWEN_Buffer_GetStart(buf));
topic=_findMaskMatchingTopic(AQHMQTT_Device_GetTopicList(newDevice), rcvdTopic, AQHMQTT_TopicDir_In);
AQHMQTT_Topic_SetTopic(topic, rcvdTopic);
if (aqh->registeredDeviceList==NULL)
aqh->registeredDeviceList=AQHMQTT_Device_List_new();
DBG_ERROR(NULL, "Registered device \"%s\" (%s)", AQHMQTT_Device_GetId(newDevice), AQHMQTT_Device_GetName(newDevice));
AQHMQTT_Device_List_Add(newDevice, aqh->registeredDeviceList);
_announceDeviceToBroker(aqh, newDevice);
GWEN_Buffer_free(buf);
return 1;
}
}
}
device=AQHMQTT_Device_List_Next(device);
}
}
DBG_INFO(AQH_LOGDOMAIN, "ignoring topic \"%s\"", rcvdTopic);
}
return 0;
}
AQHMQTT_TOPIC *_findMaskMatchingTopic(AQHMQTT_TOPIC_LIST *topicList, const char *rcvdTopic, int dir)
{
if (topicList) {
AQHMQTT_TOPIC *topic;
topic=AQHMQTT_Topic_List_First(topicList);
while(topic) {
if (AQHMQTT_Topic_GetDirection(topic)==dir) {
const char *sMask;
sMask=AQHMQTT_Topic_GetMask(topic);
if (sMask && *sMask && GWEN_Text_ComparePattern(rcvdTopic, sMask, 0)!=-1) {
/* found a matching topic */
return topic;
}
}
topic=AQHMQTT_Topic_List_Next(topic);
}
}
return NULL;
}
AQHMQTT_TOPIC *_findTopicMatchingTopic(AQHMQTT_TOPIC_LIST *topicList, const char *rcvdTopic, int dir)
{
if (topicList) {
AQHMQTT_TOPIC *topic;
topic=AQHMQTT_Topic_List_First(topicList);
while(topic) {
if (AQHMQTT_Topic_GetDirection(topic)==dir) {
const char *sTopic;
sTopic=AQHMQTT_Topic_GetTopic(topic);
if (sTopic && *sTopic && strcasecmp(rcvdTopic, sTopic)==0) {
return topic;
}
}
topic=AQHMQTT_Topic_List_Next(topic);
}
}
return NULL;
}
GWEN_BUFFER *_extractDeviceId(const AQHMQTT_TOPIC *topic, const char *rcvdTopic)
{
const char *sBefore;
const char *sAfter;
sBefore=AQHMQTT_Topic_GetBeforeId(topic);
sAfter=AQHMQTT_Topic_GetAfterId(topic);
if (sBefore && *sBefore && sAfter && *sAfter) {
GWEN_BUFFER *buf;
int rv;
buf=GWEN_Buffer_new(0, 256, 0, 1);
GWEN_Buffer_AppendString(buf, rcvdTopic);
rv=GWEN_Buffer_KeepTextBetweenStrings(buf, sBefore, sAfter, 1);
if (rv<0) {
DBG_INFO(NULL, "Could not extract id from [%s] (beforeId=%s, afterId=%s)", rcvdTopic, sBefore, sAfter);
GWEN_Buffer_free(buf);
return NULL;
}
return buf;
}
return NULL;
}

27
apps/aqhome-mqttlog/loop_mqtt.h
View File

@@ -1,27 +0,0 @@
/****************************************************************************
* This file is part of the project AqHome.
* AqHome (c) by 2023 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_LOOP_MQTT_H
#define AQHOMEMQTT_LOOP_MQTT_H
#include "./aqhome_mqtt.h"
#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/msg.h>
void AqHomeMqttLog_ReadAndHandleMqttMessages(AQHOME_MQTT *aqh);
int AqHomeMqttLog_SendPing(AQHOME_MQTT *aqh);
#endif