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

aqhome: finished transformation of aqhome-data and aqhome-tool.

Browse Source
This commit is contained in:
Martin Preuss
2025-03-02 21:48:22 +01:00
parent 2f468e4f78
commit 58c6d12e36
44 changed files with 1279 additions and 1597 deletions
Download Patch File Download Diff File Expand all files Collapse all files

81
apps/aqhome-nodes/server.c
View File

@@ -14,6 +14,7 @@
#include "./db.h"
#include "./devicesread.h"
#include "./r_setdata.h"
#include "./r_connect.h"
#include <aqhome/aqhome.h>
#include <aqhome/ipc2/ipc_endpoint.h>
@@ -27,8 +28,11 @@
#include <aqhome/ipc2/ipc_server.h>
#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/data/m_ipcd.h>
#include <aqhome/msg/ipc/data/m_ipcd_multidata.h>
#include <aqhome/msg/ipc/nodes/m_ipcn.h>
#include <aqhome/msg/ipc/nodes/m_ipcn_forward.h>
#include <aqhome/msg/node/m_node.h>
#include <aqhome/msg/node/m_value.h>
#include <aqhome/msg/node/m_recvstats.h>
@@ -94,6 +98,7 @@ static int readIntConfigWithAlt(GWEN_DB_NODE *dbArgs, const char *varName, const
static int _startIpc(AQH_OBJECT *o, AQH_NODE_SERVER *xo);
static int _startTty(AQH_OBJECT *o, AQH_NODE_SERVER *xo);
static int _startBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo);
static int _exchangeConnect(AQH_OBJECT *o, AQH_NODE_SERVER *xo, uint32_t flags);
static void _setupDb(AQH_NODE_SERVER *xo);
static int _loadDeviceList(AQH_NODE_SERVER *xo);
@@ -109,9 +114,10 @@ static void _handleMsgFromClient(AQH_OBJECT *o, AQH_NODE_SERVER *xo, AQH_OBJECT
static void _handleMsgFromTty(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _writeTtyMsgToLogFile(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardTtyMsgToBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardValueMessage(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardDataFromSendStatsMessage(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardDataFromRecvStatsMessage(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardValueMessageToBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardDataFromSendStatsMsgToBroker(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardDataFromRecvStatsMsgToBroker(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _forwardTtyMsgToClients(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg);
static void _publishInt(AQH_NODE_SERVER *xo, uint32_t uid, const char *vPath, int vModality, const char *vUnits, int v);
static void _publishDouble(AQH_NODE_SERVER *xo, uint32_t uid, const char *vPath, int vModality, const char *vUnits, double v);
static void _setDeviceName(AQH_VALUE *value, uint32_t uid);
@@ -562,14 +568,7 @@ int _startBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo)
AQH_Object_AddLink(ep, AQH_ENDPOINT_SIGNAL_CLOSED, AQH_NODE_SERVER_SLOT_BROKERCLOSED, o);
AQH_Object_Enable(ep);
rv=AQH_IpcEndpoint_ExchangeConnectMsg(ep,
AQH_MSGTYPE_IPC_DATA_CONNECT_REQ,
AQH_MSGTYPE_IPC_DATA_RESULT,
xo->brokerClientId,
NULL,
NULL,
0,
xo->timeoutInSeconds);
rv=_exchangeConnect(o, xo, 0);
if (rv!=0) {
DBG_ERROR(NULL, "Error connecting to broker: %d", rv);
return (rv<0)?rv:GWEN_ERROR_PERMISSIONS;
@@ -588,6 +587,24 @@ int _startBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo)
int _exchangeConnect(AQH_OBJECT *o, AQH_NODE_SERVER *xo, uint32_t flags)
{
AQH_MESSAGE *msgOut;
uint32_t msgId;
msgId=AQH_Endpoint_GetNextMessageId(xo->ipcEndpoint);
msgOut=AQH_IpcMessageConnect_new(xo->protoId, xo->protoVer,
AQH_MSGTYPE_IPC_CONNECT_REQ,
msgId, 0,
xo->brokerClientId, NULL, NULL, flags);
AQH_Endpoint_AddMsgOut(xo->ipcEndpoint, msgOut);
return AQH_IpcEndpoint_WaitForResultMsg(xo->ipcEndpoint,
xo->protoId, xo->protoVer, AQH_MSGTYPE_IPC_RESULT,
msgId, xo->timeoutInSeconds);
}
void _setupDb(AQH_NODE_SERVER *xo)
{
if (xo->dbFile) {
@@ -700,6 +717,7 @@ void _handleMsgFromClient(GWEN_UNUSED AQH_OBJECT *o, GWEN_UNUSED AQH_NODE_SERVER
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_CONNECT_REQ: AQH_NodeServer_HandleConnect(o, ep, msg); break;
default: break;
}
}
@@ -745,6 +763,7 @@ void _handleMsgFromTty(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *ms
AQH_NodeServer_NodeMsgToDb(o, msg);
_writeTtyMsgToLogFile(xo, msg);
_forwardTtyMsgToBroker(o, xo, msg);
_forwardTtyMsgToClients(o, xo, msg);
}
@@ -755,16 +774,44 @@ void _forwardTtyMsgToBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAG
code=AQH_NodeMessage_GetMsgType(msg);
switch(code) {
case AQH_MSG_TYPE_VALUE_REPORT: _forwardValueMessage(o, xo, msg); break;
case AQH_MSG_TYPE_COMSENDSTATS: _forwardDataFromSendStatsMessage(xo, msg); break;
case AQH_MSG_TYPE_COMRECVSTATS: _forwardDataFromRecvStatsMessage(xo, msg); break;
case AQH_MSG_TYPE_VALUE_REPORT: _forwardValueMessageToBroker(o, xo, msg); break;
case AQH_MSG_TYPE_COMSENDSTATS: _forwardDataFromSendStatsMsgToBroker(xo, msg); break;
case AQH_MSG_TYPE_COMRECVSTATS: _forwardDataFromRecvStatsMsgToBroker(xo, msg); break;
default: break;
}
}
void _forwardValueMessage(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
void _forwardTtyMsgToClients(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint8_t code;
uint32_t msgGroup;
code=AQH_NodeMessage_GetMsgType(msg);
msgGroup=AQH_NodeMessage_GetMsgGroup(code);
if (msgGroup) {
AQH_OBJECT *ep;
ep=AQH_Object_List_First(xo->ipcClientList);
while(ep) {
if (AQH_Endpoint_GetAcceptedMsgGroups(ep) & msgGroup) {
AQH_MESSAGE *outMsg;
outMsg=AQH_IpcnMessageForward_new(AQH_MSGTYPE_IPC_NODES_FORWARD,
AQH_Endpoint_GetNextMessageId(ep), 0,
AQH_Message_GetMsgPointer(msg), AQH_Message_GetUsedSize(msg));
AQH_Endpoint_AddMsgOut(ep, outMsg);
}
ep=AQH_Object_List_Next(ep);
}
}
}
void _forwardValueMessageToBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint8_t valueId;
double v;
@@ -790,7 +837,7 @@ void _forwardValueMessage(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE
void _forwardDataFromSendStatsMessage(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
void _forwardDataFromSendStatsMsgToBroker(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint16_t packetsOutInt;
@@ -820,7 +867,7 @@ void _forwardDataFromSendStatsMessage(AQH_NODE_SERVER *xo, const AQH_MESSAGE *ms
void _forwardDataFromRecvStatsMessage(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
void _forwardDataFromRecvStatsMsgToBroker(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint16_t packetsInInt;