450 lines
13 KiB
C
450 lines
13 KiB
C
/****************************************************************************
|
|
* 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_publish.h"
|
|
#include "./server_p.h"
|
|
|
|
#include <aqhome/aqhome.h>
|
|
#include <aqhome/msg/ipc/data/m_ipcd.h>
|
|
#include <aqhome/msg/ipc/data/m_ipcd_multidata.h>
|
|
#include <aqhome/msg/ipc/data/m_ipcd_values.h>
|
|
#include <aqhome/ipc2/endpoint.h>
|
|
#include <aqhome/msg/mqtt/m_mqtt_publish.h>
|
|
|
|
#include <gwenhywfar/json_read.h>
|
|
#include <gwenhywfar/text.h>
|
|
#include <gwenhywfar/debug.h>
|
|
|
|
|
|
|
|
/* ------------------------------------------------------------------------------------------------
|
|
* forward declarations
|
|
* ------------------------------------------------------------------------------------------------
|
|
*/
|
|
|
|
static int _handlePublish(AQH_OBJECT *o, const char *rcvdTopic, const char *rcvdValue);
|
|
static void _handleNumTopic(AQH_MQTTLOG_SERVER *xo, AQHMQTT_DEVICE *device,
|
|
AQHMQTT_TOPIC *topic, const char *rcvdValue);
|
|
static void _handleJsonTopic(AQH_MQTTLOG_SERVER *xo, AQHMQTT_DEVICE *device,
|
|
AQHMQTT_TOPIC *topic, const char *rcvdValue);
|
|
static void _sendMessage(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value, const char *rcvdValue);
|
|
static void _announceDeviceToBroker(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *device);
|
|
static void _sendAnnounceValueMessage(AQH_MQTTLOG_SERVER *xo, 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(AQH_MQTTLOG_SERVER *xo, const char *rcvdTopic);
|
|
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);
|
|
|
|
|
|
|
|
/* ------------------------------------------------------------------------------------------------
|
|
* code
|
|
* ------------------------------------------------------------------------------------------------
|
|
*/
|
|
|
|
void AQH_MqttLogServer_HandlePublishMsg(AQH_OBJECT *o, GWEN_UNUSED AQH_OBJECT *ep, const AQH_MESSAGE *msg)
|
|
{
|
|
if (o && msg) {
|
|
AQH_MQTTLOG_SERVER *xo;
|
|
|
|
xo=AQH_MqttLogServer_GetServerData(o);
|
|
if (xo && xo->registeredDeviceList) {
|
|
char *topic;
|
|
char *value;
|
|
|
|
topic=AQH_MqttMessagePublish_ExtractTopic(msg);
|
|
value=AQH_MqttMessagePublish_ExtractValue(msg);
|
|
|
|
if (topic && value) {
|
|
int rv;
|
|
|
|
rv=_handlePublish(o, topic, value);
|
|
if (rv!=1) {
|
|
DBG_INFO(NULL, "New topic \"%s\", trying to register", topic);
|
|
rv=_registerNewDeviceForTopic(xo, topic);
|
|
if (rv==1) {
|
|
rv=_handlePublish(o, 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(AQH_OBJECT *o, const char *rcvdTopic, const char *rcvdValue)
|
|
{
|
|
if (o && rcvdTopic && *rcvdTopic) {
|
|
AQH_MQTTLOG_SERVER *xo;
|
|
|
|
xo=AQH_MqttLogServer_GetServerData(o);
|
|
if (xo && xo->registeredDeviceList) {
|
|
AQHMQTT_DEVICE *device;
|
|
|
|
device=AQHMQTT_Device_List_First(xo->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(xo, device, topic, rcvdValue);
|
|
else
|
|
_handleNumTopic(xo, device, topic, rcvdValue);
|
|
return 1;
|
|
}
|
|
}
|
|
|
|
device=AQHMQTT_Device_List_Next(device);
|
|
}
|
|
DBG_INFO(AQH_LOGDOMAIN, "ignoring topic \"%s\"", rcvdTopic);
|
|
}
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
|
|
|
|
void _handleNumTopic(AQH_MQTTLOG_SERVER *xo, AQHMQTT_DEVICE *device,
|
|
AQHMQTT_TOPIC *topic, const char *rcvdValue)
|
|
{
|
|
AQHMQTT_VALUE_LIST *valueList;
|
|
|
|
valueList=AQHMQTT_Topic_GetValueList(topic);
|
|
if (valueList)
|
|
_sendMessage(xo, device, AQHMQTT_Value_List_First(valueList), rcvdValue);
|
|
else {
|
|
DBG_INFO(NULL, "No value list in device \"%s\"", AQHMQTT_Device_GetId(device));
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void _handleJsonTopic(AQH_MQTTLOG_SERVER *xo, 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(xo, device, value, s);
|
|
}
|
|
}
|
|
value=AQHMQTT_Value_List_Next(value);
|
|
} /* while */
|
|
}
|
|
GWEN_JsonElement_free(jeRoot);
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void _sendMessage(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value, const char *rcvdValue)
|
|
{
|
|
int rv;
|
|
double f;
|
|
const char *deviceName;
|
|
|
|
deviceName=AQHMQTT_Device_GetId(device);
|
|
rv=GWEN_Text_StringToDouble(rcvdValue, &f);
|
|
if (rv<0) {
|
|
DBG_ERROR(NULL, "Invalid value received from MQTT server (%s)", rcvdValue?rcvdValue:"<empty>");
|
|
}
|
|
else {
|
|
uint64_t now;
|
|
AQH_VALUE *msgValue;
|
|
|
|
now=(uint64_t) time(NULL);
|
|
msgValue=_mkMessageValue(device, value);
|
|
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);
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void _announceDeviceToBroker(AQH_MQTTLOG_SERVER *xo, 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(xo, device, value);
|
|
value=AQHMQTT_Value_List_Next(value);
|
|
}
|
|
}
|
|
|
|
topic=AQHMQTT_Topic_List_Next(topic);
|
|
}
|
|
}
|
|
}
|
|
|
|
|
|
|
|
void _sendAnnounceValueMessage(AQH_MQTTLOG_SERVER *xo, const AQHMQTT_DEVICE *device, const AQHMQTT_VALUE *value)
|
|
{
|
|
AQH_MESSAGE *pubMsg;
|
|
AQH_VALUE *msgValue;
|
|
|
|
msgValue=_mkMessageValue(device, value);
|
|
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 *_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(AQH_MQTTLOG_SERVER *xo, const char *rcvdTopic)
|
|
{
|
|
if (rcvdTopic && *rcvdTopic) {
|
|
if (xo->availableDeviceList) {
|
|
AQHMQTT_DEVICE *device;
|
|
|
|
device=AQHMQTT_Device_List_First(xo->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 (xo->registeredDeviceList==NULL)
|
|
xo->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, xo->registeredDeviceList);
|
|
_announceDeviceToBroker(xo, 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;
|
|
}
|
|
|
|
|
|
|