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

Revert "Revert "aqhome: convenience code.""

Browse Source 
This reverts commit bb77c6acd1.
This commit is contained in:
Martin Preuss
2024-09-30 18:28:38 +02:00
parent 0566483575
commit 03f9178dd2
5 changed files with 435 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

78
aqhome/client/0BUILD Normal file
View File

@@ -0,0 +1,78 @@
<?xml?>
<gwbuild>
<target type="ConvenienceLibrary" name="aqhclient" >
<includes type="c" >
$(gwenhywfar_cflags)
-I$(topsrcdir)
-I$(topbuilddir)
</includes>
<includes type="tm2" >
--include=$(builddir)
--include=$(srcdir)
</includes>
<define name="BUILDING_AQHOME" />
<setVar name="local/cflags">$(visibility_cflags)</setVar>
<setVar name="tm2flags" >
--api=AQHOME_API
</setVar>
<setVar name="local/typefiles" >
</setVar>
<setVar name="local/built_sources" >
</setVar>
<setVar name="local/built_headers_pub">
</setVar>
<setVar name="local/built_headers_priv" >
</setVar>
<headers dist="false" install="$(pkgincludedir)/client" >
$(local/built_headers_pub)
</headers>
<headers dist="true" install="$(pkgincludedir)/client" >
connection.h
</headers>
<headers dist="true" >
</headers>
<sources>
$(local/typefiles)
connection.c
</sources>
<extradist>
</extradist>
<useTargets>
</useTargets>
<subdirs>
</subdirs>
</target>
</gwbuild>

175
aqhome/client/connection.c Normal file
View File

@@ -0,0 +1,175 @@
/****************************************************************************
* 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 "./connection.h"
#include "aqhome/ipc/endpoint_ipc.h"
#include "aqhome/ipc/endpoint_ipcclient.h"
#include "aqhome/ipc/data/ipc_data.h"
#include "aqhome/ipc/data/msg_data_connect.h"
#include "aqhome/ipc/msg_ipc_result.h"
#include <gwenhywfar/endpoint_tcpc.h>
#include <gwenhywfar/endpoint_multilayer.h>
#include <gwenhywfar/msg_ipc.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/timestamp.h>
#include <gwenhywfar/db.h>
#include <time.h>
GWEN_MSG_ENDPOINT *_physConnectToBroker(const char *addr, int port, const char *clientId, uint32_t flags);
int AQH_BrokerConnection_FlushOutMessageQueue(GWEN_MSG_ENDPOINT *epTcp, int timeoutInSeconds)
{
time_t startTime;
startTime=time(NULL);
while(GWEN_MsgEndpoint_HaveMessageToSend(epTcp)) {
time_t now;
GWEN_MsgEndpoint_IoLoop(epTcp, 1000); /* 1000 ms */
now=time(NULL);
if (now-startTime>timeoutInSeconds) {
DBG_INFO(AQH_LOGDOMAIN, "Timeout");
return GWEN_ERROR_TIMEOUT;
}
}
return 0;
}
GWEN_MSG *AQH_BrokerConnection_WaitForSpecificMessage(GWEN_MSG_ENDPOINT *epTcp, int msgCode, int timeoutInSeconds)
{
time_t startTime;
startTime=time(NULL);
for (;;) {
GWEN_MSG *msg;
time_t now;
GWEN_MsgEndpoint_IoLoop(epTcp, 1000); /* 1000 ms */
msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp);
if (msg) {
uint16_t code;
code=GWEN_IpcMsg_GetCode(msg);
if (code==msgCode) {
DBG_INFO(AQH_LOGDOMAIN, "Received expected IPC message");
return msg;
}
else if (code==AQH_MSGTYPE_IPC_DATA_RESULT) {
DBG_INFO(AQH_LOGDOMAIN, "Received IPC result message");
return msg;
}
else {
DBG_INFO(AQH_LOGDOMAIN, "Received unexpected message %d (%x)", code, code);
}
GWEN_Msg_free(msg);
}
now=time(NULL);
if (now-startTime>timeoutInSeconds) {
DBG_INFO(AQH_LOGDOMAIN, "Timeout");
break;
}
}
return NULL;
}
GWEN_MSG_ENDPOINT *AQH_BrokerConnection_OpenConnection(const char *addr, int port,
const char *clientId,
const char *userId, const char *passwd,
uint32_t flags,
int timeoutInSeconds)
{
GWEN_MSG_ENDPOINT *epTcp;
GWEN_MSG *msgOut;
GWEN_MSG *msgIn;
uint32_t result;
epTcp=_physConnectToBroker(addr, port, clientId, 0);
if (epTcp==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "ERROR creating TCP connection");
return NULL;
}
msgOut=AQH_ConnectDataIpcMsg_new(AQH_MSGTYPE_IPC_DATA_CONNECT_REQ,
GWEN_MsgEndpoint_GetNextMessageId(epTcp), 0,
clientId, userId, passwd, flags);
if (msgOut==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "Error creating message");
GWEN_MsgEndpoint_free(epTcp);
return NULL;
}
GWEN_MsgEndpoint_AddSendMessage(epTcp, msgOut);
msgIn=AQH_BrokerConnection_WaitForSpecificMessage(epTcp, AQH_MSGTYPE_IPC_DATA_RESULT, timeoutInSeconds);
if (msgIn==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "No response received");
GWEN_MsgEndpoint_free(epTcp);
return NULL;
}
result=AQH_ResultIpcMsg_GetResultCode(msgIn);
GWEN_Msg_free(msgIn);
if (result!=AQH_MSG_IPC_SUCCESS) {
DBG_ERROR(AQH_LOGDOMAIN, "Response: %d", result);
GWEN_MsgEndpoint_free(epTcp);
return NULL;
}
return epTcp;
}
GWEN_MSG_ENDPOINT *_physConnectToBroker(const char *addr, int port, const char *clientId, uint32_t flags)
{
GWEN_MSG_ENDPOINT *ep;
GWEN_MSG_ENDPOINT *ipcBaseEndpoint;
int rv;
ep=AQH_ClientIpcEndpoint_new("brokerIpcClient", 0);
GWEN_MsgEndpoint_AddFlags(ep, flags);
ipcBaseEndpoint=AQH_IpcEndpoint_CreateIpcTcpClient(addr, port, "brokerPhysEndpoint", 0);
AQH_IpcEndpoint_SetServiceName(ipcBaseEndpoint, clientId);
GWEN_MsgEndpoint_Tree2_AddChild(ep, ipcBaseEndpoint);
rv=GWEN_MultilayerEndpoint_StartConnect(ep);
if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) {
DBG_ERROR(AQH_LOGDOMAIN, "Error connecting to broker server %s:%d (%d)", addr, port, rv);
GWEN_MsgEndpoint_free(ep);
return NULL;
}
return ep;
}

28
aqhome/client/connection.h Normal file
View File

@@ -0,0 +1,28 @@
/****************************************************************************
* 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.
****************************************************************************/
#ifndef AQHOME_CLIENT_CONNECTION_H
#define AQHOME_CLIENT_CONNECTION_H
#include <aqhome/api.h>
#include <gwenhywfar/endpoint.h>
GWEN_MSG_ENDPOINT *AQH_BrokerConnection_OpenConnection(const char *addr, int port,
const char *clientId,
const char *userId, const char *passwd,
uint32_t flags,
int timeoutInSeconds);
int AQH_BrokerConnection_FlushOutMessageQueue(GWEN_MSG_ENDPOINT *epTcp, int timeoutInSeconds);
GWEN_MSG *AQH_BrokerConnection_WaitForSpecificMessage(GWEN_MSG_ENDPOINT *epTcp, int msgCode, int timeoutInSeconds);
#endif

44
aqhome/ipc/data/cmd_data.c Normal file
View File

@@ -0,0 +1,44 @@
GWEN_MSG *AQH_CmdDataIpc_WaitForSpecificIpcMessage(GWEN_MSG_ENDPOINT *epTcp, int msgCode, int timeoutInSeconds)
{
time_t startTime;
startTime=time(NULL);
for (;;) {
GWEN_MSG *msg;
time_t now;
while( (msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp)) ) {
uint16_t code;
code=GWEN_IpcMsg_GetCode(msg);
if (code==msgCode) {
DBG_INFO(NULL, "Received expected IPC message");
return msg;
}
else if (code==AQH_MSGTYPE_IPC_DATA_RESULT) {
DBG_INFO(NULL, "Received IPC result message");
return msg;
}
else {
DBG_INFO(NULL, "Received unexpected message %d (%x)", code, code);
}
GWEN_Msg_free(msg);
}
now=time(NULL);
if (now-startTime>timeoutInSeconds) {
DBG_INFO(NULL, "Timeout");
break;
}
GWEN_MsgEndpoint_IoLoop(epTcp, 2000); /* 2000 ms */
}
return NULL;
}

110
aqhome/ipc/data/cmd_data_getdata.c Normal file
View File

@@ -0,0 +1,110 @@
/****************************************************************************
* 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/ipc/data/msg_data_datapoints.h>
#include <aqhome/ipc/data/ipc_data.h>
#include <aqhome/ipc/data/msg_data_multidata.h>
#include <aqhome/ipc/data/msg_data_getdata.h>
#include <aqhome/ipc/data/ipc_data.h>
#include <aqhome/ipc/msg_ipc_tag16.h>
#include <gwenhywfar/msg.h>
#include <gwenhywfar/buffer.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/msg_ipc.h>
GWEN_MSG *_waitForMatchingResponse(GWEN_MSG_ENDPOINT *epTcp, int msgCode, int timeoutInSeconds);
GWEN_MSG *AQH_CmdDataIpc_RequestDatapointsMsg(GWEN_MSG_ENDPOINT *ep,
const char *valueName, uint64_t tsBegin, uint64_t tsEnd, uint64_t num,
int timeoutInSeconds)
{
GWEN_MSG *msg;
time_t startTime;
msg=AQH_GetDataDataIpcMsg_new(AQH_MSGTYPE_IPC_DATA_GETDATA_REQ, valueName, tsBegin, tsEnd, num);
GWEN_MsgEndpoint_AddSendMessage(ep, msg);
msg=_waitForMatchingResponse(ep, AQH_MSGTYPE_IPC_DATA_GETDATA_RSP, timeoutInSeconds);
if (msg) {
uint16_t code;
code=GWEN_IpcMsg_GetCode(msg);
if (code==AQH_MSGTYPE_IPC_DATA_RESULT) {
uint32_t resultCode;
resultCode=AQH_ResultIpcMsg_GetResultCode(msg);
DBG_INFO(AQH_LOGDOMAIN, "IPC error: %d", resultCode);
GWEN_Msg_free(msg);
return NULL;
}
else if (code==AQH_MSGTYPE_IPC_DATA_GETDATA_RSP) {
int rv;
AQH_MultiDataDataIpcMsg_Parse(msg, 0);
return msg;
}
GWEN_Msg_free(msg);
}
return NULL;
}
GWEN_MSG *_waitForMatchingResponse(GWEN_MSG_ENDPOINT *epTcp, int msgCode, int timeoutInSeconds)
{
time_t startTime;
startTime=time(NULL);
for (;;) {
GWEN_MSG *msg;
time_t now;
while( (msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp)) ) {
uint16_t code;
code=GWEN_IpcMsg_GetCode(msg);
if (code==msgCode) {
DBG_INFO(NULL, "Received expected IPC message");
return msg;
}
else if (code==AQH_MSGTYPE_IPC_DATA_RESULT) {
DBG_INFO(NULL, "Received IPC result message");
return msg;
}
else {
DBG_INFO(NULL, "Received unexpected message %d (%x), ignoring", code, code);
}
GWEN_Msg_free(msg);
} /* while */
now=time(NULL);
if (now-startTime>timeoutInSeconds) {
DBG_INFO(NULL, "Timeout");
break;
}
GWEN_MsgEndpoint_IoLoop(epTcp, 2000); /* 2000 ms */
}
return NULL;
}