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

Heavy work on IPC.

Browse Source 
We will now have a broker (aqhome-data) which stores data and distributes
value change messages among connected clients.
aqhomed will connect to that broker and send its values there.
aqhome-mqtt will also connect to the broker and send its values there.
Other clients can later connect to check for changes and react according
to rules.
This commit is contained in:
Martin Preuss
2023-09-10 23:13:03 +02:00
parent 2b733a52ca
commit 518a3a53f9
43 changed files with 1412 additions and 707 deletions
Download Patch File Download Diff File Expand all files Collapse all files

271
apps/aqhome-data/loop.c
View File

@@ -12,6 +12,11 @@
#include "./loop.h"
#include "./c_connect.h"
#include "./c_updatedata.h"
#include "./c_getdatapoints.h"
#include "./c_getlastdatapoint.h"
#include "./c_getvalues.h"
#include "./aqhome_data_p.h"
#include "aqhome/ipc/data/ipc_data.h"
#include "aqhome/ipc/data/msg_data_values.h"
@@ -31,8 +36,6 @@
* ------------------------------------------------------------------------------------------------
*/
#define AQHOMEDATA_VALUESPERMSG 10
/* ------------------------------------------------------------------------------------------------
@@ -44,14 +47,6 @@ static void _readAndHandleIpcMessages(AQHOME_DATA *aqh);
static void _handleIpcEndpoint(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep);
static void _handleIpcMsg(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _handleGetValues(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _sendValueList(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE_LIST *vl, uint32_t flags);
static void _handleAddValue(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _handleEditValues(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _handleAddDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _handleGetDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
static void _handleGetLastDataPoint(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg);
/* ------------------------------------------------------------------------------------------------
@@ -134,260 +129,14 @@ void _handleIpcMsg(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
code=GWEN_IpcMsg_GetCode(msg);
DBG_ERROR(AQH_LOGDOMAIN, "Received IPC packet %d", (int) code);
switch(code) {
case AQH_MSGTYPE_IPC_DATA_GETVALUES_REQ: _handleGetValues(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_ADDVALUES_REQ: _handleAddValue(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_EDITVALUE_REQ: _handleEditValues(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_ADDDATAPOINTS_REQ: _handleAddDataPoints(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_GETDATAPOINTS_REQ: _handleGetDataPoints(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_GETLASTDATAPOINT_REQ: _handleGetLastDataPoint(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_CONNECT_REQ: AqHomeData_HandleConnect(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_UPDATEDATA: AqHomeData_HandleUpdateData(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_GETVALUES_REQ: AqHomeData_HandleGetValues(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_GETDATA_REQ: AqHomeData_HandleGetDataPoints(aqh, ep, msg); break;
case AQH_MSGTYPE_IPC_DATA_GETLASTDATAPOINT_REQ: AqHomeData_HandleGetLastDataPoint(aqh, ep, msg); break;
default: break;
}
}
void _handleGetValues(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
{
const AQH_VALUE_LIST *origValueList;
origValueList=AQH_Storage_GetValueList(aqh->storage);
if (origValueList) {
if (AQH_Value_List_GetCount(origValueList)<AQHOMEDATA_VALUESPERMSG)
_sendValueList(aqh, ep, origValueList, AQH_MSGDATA_VALUES_FLAGS_LASTMSG);
else {
AQH_VALUE_LIST *tmpValueList;
const AQH_VALUE *v;
tmpValueList=AQH_Value_List_new();
v=AQH_Value_List_First(origValueList);
while(v) {
const AQH_VALUE *next;
AQH_VALUE *copyOfValue;
next=AQH_Value_List_Next(v);
copyOfValue=AQH_Value_dup(v);
AQH_Value_List_Add(copyOfValue, tmpValueList);
if (AQH_Value_List_GetCount(tmpValueList)>=AQHOMEDATA_VALUESPERMSG) {
_sendValueList(aqh, ep, tmpValueList, next?0:AQH_MSGDATA_VALUES_FLAGS_LASTMSG);
AQH_Value_List_Clear(tmpValueList);
}
v=next;
}
if (AQH_Value_List_GetCount(tmpValueList))
_sendValueList(aqh, ep, tmpValueList, AQH_MSGDATA_VALUES_FLAGS_LASTMSG); /* send remaining */
AQH_Value_List_free(tmpValueList);
}
}
else {
/* empty list */
_sendValueList(aqh, ep, NULL, AQH_MSGDATA_VALUES_FLAGS_LASTMSG);
}
}
void _sendValueList(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const AQH_VALUE_LIST *vl, uint32_t flags)
{
GWEN_MSG *msg;
msg=AQH_ValuesDataIpcMsg_new(AQH_MSGTYPE_IPC_DATA_GETVALUES_RSP, flags, vl);
GWEN_MsgEndpoint_AddSendMessage(ep, msg);
}
void _handleAddValue(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *recvdMsg)
{
GWEN_MSG *outMsg;
int resultCode=0;
if (AQH_ValuesDataIpcMsg_IsValid(recvdMsg)) {
uint32_t numValues;
numValues=AQH_ValuesDataIpcMsg_GetNumValues(recvdMsg);
if (numValues==1) {
const char *s;
s=AQH_ValuesDataIpcMsg_GetValueName(recvdMsg, 0);
if (s && *s) {
if (AQH_Storage_GetValueByName(aqh->storage, s)==NULL) {
AQH_VALUE *v;
v=AQH_Value_new();
AQH_Value_SetName(v, s);
s=AQH_ValuesDataIpcMsg_GetValueUnits(recvdMsg, 0);
if (s && *s)
AQH_Value_SetValueUnits(v, s);
DBG_INFO(NULL, "Adding value \"%s\" (%s)", AQH_Value_GetName(v), AQH_Value_GetValueUnits(v));
AQH_Storage_AddValue(aqh->storage, v);
AQH_Storage_AddRuntimeFlags(aqh->storage, AQH_STORAGE_RTFLAGS_MODIFIED);
resultCode=AQH_MSG_IPC_SUCCESS;
}
else {
DBG_INFO(NULL, "Value \"%s\" already exists", s);
resultCode=AQH_MSG_IPC_ERROR_EXISTS;
}
}
else {
DBG_INFO(NULL, "Value without name ");
resultCode=AQH_MSG_IPC_ERROR_INVALID;
}
}
else {
DBG_INFO(NULL, "Invalid number of values in message (%d)", numValues);
resultCode=AQH_MSG_IPC_ERROR_INVALID;
}
}
else {
DBG_INFO(NULL, "Invalid message received");
resultCode=AQH_MSG_IPC_ERROR_BADDATA;
}
outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, resultCode);
GWEN_MsgEndpoint_AddSendMessage(ep, outMsg);
}
void _handleEditValues(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *recvdMsg)
{
GWEN_MSG *outMsg;
int resultCode=0;
if (AQH_ValuesDataIpcMsg_IsValid(recvdMsg)) {
uint32_t numValues;
numValues=AQH_ValuesDataIpcMsg_GetNumValues(recvdMsg);
if (numValues==1) {
const char *s;
s=AQH_ValuesDataIpcMsg_GetValueName(recvdMsg, 0);
if (s && *s) {
AQH_VALUE *v;
v=AQH_Storage_GetValueByName(aqh->storage, s);
if (v==NULL) {
DBG_INFO(NULL, "Value \"%s\" doesn't not exist", s);
resultCode=AQH_MSG_IPC_ERROR_EXISTS;
}
else {
DBG_INFO(NULL, "Updating value \"%s\" (%s)", AQH_Value_GetName(v), AQH_Value_GetValueUnits(v));
s=AQH_ValuesDataIpcMsg_GetValueUnits(recvdMsg, 0);
if (s && *s)
AQH_Value_SetValueUnits(v, s);
AQH_Storage_AddRuntimeFlags(aqh->storage, AQH_STORAGE_RTFLAGS_MODIFIED);
resultCode=AQH_MSG_IPC_SUCCESS;
}
}
else {
DBG_INFO(NULL, "Value without name ");
resultCode=AQH_MSG_IPC_ERROR_INVALID;
}
}
else {
DBG_INFO(NULL, "Invalid number of values in message (%d)", numValues);
resultCode=AQH_MSG_IPC_ERROR_INVALID;
}
}
else {
DBG_INFO(NULL, "Invalid message received");
resultCode=AQH_MSG_IPC_ERROR_BADDATA;
}
outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, resultCode);
GWEN_MsgEndpoint_AddSendMessage(ep, outMsg);
}
void _handleAddDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *recvdMsg)
{
GWEN_MSG *outMsg;
int resultCode=0;
if (AQH_DataPointsDataIpcMsg_IsValid(recvdMsg)) {
uint32_t numValues;
numValues=AQH_DataPointsDataIpcMsg_GetNumValues(recvdMsg);
if (numValues) {
const char *s;
s=AQH_DataPointsDataIpcMsg_GetValueName(recvdMsg);
if (s && *s) {
AQH_VALUE *v;
v=AQH_Storage_GetValueByName(aqh->storage, s);
if (v==NULL) {
// TODO: maybe create the value on the fly
DBG_INFO(NULL, "Value \"%s\" doesn't not exist", s);
resultCode=AQH_MSG_IPC_ERROR_EXISTS;
}
else {
const uint64_t *dataPoints;
dataPoints=AQH_DataPointsDataIpcMsg_GetDataPoints(recvdMsg);
if (dataPoints) {
uint32_t i;
for(i=0; i<numValues; i++) {
uint64_t timestamp;
union {double f; uint64_t i;} u;
int rv;
timestamp=*(dataPoints++);
u.i=*(dataPoints++);
rv=AQH_Storage_AddDatapoint(aqh->storage, AQH_Value_GetId(v), timestamp, u.f);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
resultCode=AQH_MSG_IPC_ERROR_GENERIC;
}
else {
DBG_INFO(NULL, "Datapoint added for value \"%s\"", s);
resultCode=0;
}
} /* for */
} /* if datapoints */
else {
DBG_INFO(NULL, "No datapoints");
resultCode=AQH_MSG_IPC_ERROR_BADDATA;
}
}
}
else {
DBG_INFO(NULL, "Value without name ");
resultCode=AQH_MSG_IPC_ERROR_INVALID;
}
}
else {
DBG_INFO(NULL, "No datapoints");
resultCode=AQH_MSG_IPC_ERROR_BADDATA;
}
}
else {
DBG_INFO(NULL, "Invalid message received");
resultCode=AQH_MSG_IPC_ERROR_BADDATA;
}
outMsg=AQH_ResultIpcMsg_new(AQH_MSGTYPE_IPC_DATA_RESULT, resultCode);
GWEN_MsgEndpoint_AddSendMessage(ep, outMsg);
}
void _handleGetDataPoints(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
{
}
void _handleGetLastDataPoint(AQHOME_DATA *aqh, GWEN_MSG_ENDPOINT *ep, const GWEN_MSG *msg)
{
}