martin/aqhomecontrol
Public Access
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
6bb499740d9350807357df0bee331ce38bdf4df7
aqhomecontrol/apps/aqhome-nodes/server.c
Martin Preuss 3f2ae9dd03 Revert "started working on mqtt support in aqhome-nodes" 
This reverts commit 7fbc616ce4.
2025-08-03 00:56:46 +02:00

1449 lines
40 KiB
C
Raw Blame History

/****************************************************************************
* 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 "./server_p.h"
#include "./db.h"
#include "./devicesread.h"
#include "./r_setdata.h"
#include "./r_connect.h"
#include "./r_forward.h"
#include "./r_setaccmsggrps.h"
#include "./r_getnodes.h"
#include <aqhome/aqhome.h>
#include <aqhome/ipc2/ipc_endpoint.h>
#include <aqhome/ipc2/ipc_server.h>
#include <aqhome/ipc2/tcpd_object.h>
#include <aqhome/ipc2/tty_endpoint.h>
#include <aqhome/ipc2/tty_object.h>
#include <aqhome/ipc2/tcp_object.h>
#include <aqhome/ipc2/tcpd_object.h>
#include <aqhome/ipc2/ipc_client.h>
#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/m_ipc_tag16.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>
#include <aqhome/msg/node/m_sendstats.h>
#include <aqhome/data/value.h>
#include <gwenhywfar/args.h>
#include <gwenhywfar/misc.h>
#include <gwenhywfar/gwentime.h>
#include <gwenhywfar/debug.h>
#include <fcntl.h>
#include <unistd.h>
#include <errno.h>
/* ------------------------------------------------------------------------------------------------
* defines
* ------------------------------------------------------------------------------------------------
*/
#define I18N(msg) msg
#define I18S(msg) msg
#define A_ARG GWEN_ARGS_FLAGS_HAS_ARGUMENT
#define A_END (GWEN_ARGS_FLAGS_HELP | GWEN_ARGS_FLAGS_LAST)
#define A_CHAR GWEN_ArgsType_Char
#define A_INT GWEN_ArgsType_Int
#define AQH_NODE_SERVER_BROKER_RESTARTTIME 10
#define AQH_NODE_SERVER_TTY_RESTARTTIME 10
enum {
AQH_NODE_SERVER_SLOT_NEWCLIENT=1,
AQH_NODE_SERVER_SLOT_CLIENTCLOSED,
AQH_NODE_SERVER_SLOT_BROKERCLOSED,
AQH_NODE_SERVER_SLOT_TTYCLOSED
};
/* ------------------------------------------------------------------------------------------------
* global vars
* ------------------------------------------------------------------------------------------------
*/
GWEN_INHERIT(AQH_OBJECT, AQH_NODE_SERVER)
/* ------------------------------------------------------------------------------------------------
* forward declarations
* ------------------------------------------------------------------------------------------------
*/
static void GWENHYWFAR_CB _freeData(void *bp, void *p);
static void _readConfig(AQH_OBJECT *o, AQH_NODE_SERVER *xo, GWEN_DB_NODE *dbArgs);
static const char *readCharConfigWithAlt(GWEN_DB_NODE *dbArgs, const char *varName, const char *altVarName, const char *defaultValue);
static int readIntConfigWithAlt(GWEN_DB_NODE *dbArgs, const char *varName, const char *altVarName, int defaultValue, int nonValue);
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_NODE_SERVER *xo, uint32_t flags);
static void _setupDb(AQH_NODE_SERVER *xo);
static int _loadDeviceList(AQH_NODE_SERVER *xo);
static int _handleSignal(AQH_OBJECT *o, uint32_t slotId, AQH_OBJECT *senderObject, int param1, void *param2);
static int _handleNewIpcClient(AQH_OBJECT *o, AQH_NODE_SERVER *xo, AQH_OBJECT *clientEndpoint);
static int _handleIpcClientDown(AQH_OBJECT *clientEndpoint);
static int _handleBrokerDown(AQH_NODE_SERVER *xo);
static int _handleTtyDown(AQH_NODE_SERVER *xo);
static void _handleMsgsFromClient(AQH_OBJECT *o, AQH_NODE_SERVER *xo, AQH_OBJECT *ep);
static void _handleMsgFromClient(AQH_OBJECT *o, AQH_NODE_SERVER *xo, AQH_OBJECT *ep, const AQH_MESSAGE *msg);
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 _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_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);
static void _handleMsgFromBroker(AQH_OBJECT *o, AQH_OBJECT *ep, const AQH_MESSAGE *msg);
static void _writeToLogFile(const char *filename, const char *txt);
static int _createPidFile(const char *pidFilename);
static int _diffInSeconds(time_t t1, time_t t0);
static int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs);
/* ------------------------------------------------------------------------------------------------
* code
* ------------------------------------------------------------------------------------------------
*/
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* constructor, destructor
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
AQH_OBJECT *AQH_NodeServer_new(AQH_EVENT_LOOP *eventLoop)
{
AQH_OBJECT *o;
AQH_NODE_SERVER *xo;
o=AQH_Object_new(eventLoop);
GWEN_NEW_OBJECT(AQH_NODE_SERVER, xo);
GWEN_INHERIT_SETDATA(AQH_OBJECT, AQH_NODE_SERVER, o, xo, _freeData);
xo->ipcClientList=AQH_Object_List_new();
xo->requestTree=AQH_MsgRequest_new();
AQH_Object_SetSignalHandlerFn(o, _handleSignal);
return o;
}
void GWENHYWFAR_CB _freeData(GWEN_UNUSED void *bp, void *p)
{
AQH_NODE_SERVER *xo;
xo=(AQH_NODE_SERVER*) p;
if (xo->ipcClientList) {
AQH_Object_List_free(xo->ipcClientList);
xo->ipcClientList=NULL;
}
AQH_Object_free(xo->ipcEndpoint);
AQH_Object_free(xo->ttyEndpoint);
AQH_Object_free(xo->brokerEndpoint);
AQHNODE_Device_List_free(xo->deviceDefList);
GWEN_DB_Group_free(xo->dbArgs);
AQH_MsgRequest_free(xo->requestTree);
AQH_NodeDb_free(xo->nodeDb);
free(xo->dbFile);
free(xo->logFile);
free(xo->pidFile);
free(xo->devicePath);
free(xo->tcpAddress);
free(xo->brokerAddress);
free(xo->brokerClientId);
GWEN_FREE_OBJECT(xo);
}
AQH_NODE_SERVER *AQH_NodeServer_GetServerData(const AQH_OBJECT *o)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
return xo;
}
return NULL;
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* getter, setter
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
int AQH_NodeServer_GetTimeout(const AQH_OBJECT *o)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo)
return xo->timeout;
}
return 0;
}
void AQH_NodeServer_SetLogFile(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->logFile);
xo->logFile=s?strdup(s):NULL;
}
}
}
void AQH_NodeServer_SetDbFile(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->dbFile);
xo->dbFile=s?strdup(s):NULL;
}
}
}
void AQH_NodeServer_SetPidFile(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->pidFile);
xo->pidFile=s?strdup(s):NULL;
}
}
}
void AQH_NodeServer_SetDevicePath(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->devicePath);
xo->devicePath=s?strdup(s):NULL;
}
}
}
void AQH_NodeServer_SetTpcAddress(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->tcpAddress);
xo->tcpAddress=s?strdup(s):NULL;
}
}
}
void AQH_NodeServer_SetTcpPort(AQH_OBJECT *o, int i)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo)
xo->tcpPort=i;
}
}
void AQH_NodeServer_SetBrokerAddress(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->brokerAddress);
xo->brokerAddress=s?strdup(s):NULL;
}
}
}
void AQH_NodeServer_SetBrokerPort(AQH_OBJECT *o, int i)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo)
xo->brokerPort=i;
}
}
void AQH_NodeServer_SetBrokerClientId(AQH_OBJECT *o, const char *s)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
free(xo->brokerClientId);
xo->brokerClientId=s?strdup(s):NULL;
}
}
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* init
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
int AQH_NodeServer_Init(AQH_OBJECT *o, int argc, char **argv)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
GWEN_DB_NODE *dbArgs;
int rv;
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");
_readConfig(o, xo, 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);
}
s=GWEN_DB_GetCharValue(dbArgs, "pidfile", 0, AQHOMED_DEFAULT_PIDFILE);
if (s && *s) {
AQH_NodeServer_SetPidFile(o, s);
rv=_createPidFile(s);
if (rv<0) {
DBG_ERROR(NULL, "Error creating PID file (%d)", rv);
return rv;
}
}
DBG_INFO(NULL, "Loading device files");
rv=_loadDeviceList(xo);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
DBG_INFO(NULL, "Setup node db");
_setupDb(xo);
DBG_INFO(NULL, "Starting IPC Service");
rv=_startIpc(o, xo);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
DBG_INFO(NULL, "Starting TTY Service");
rv=_startTty(o, xo);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
DBG_INFO(NULL, "Starting Broker Connection");
rv=_startBroker(o, xo);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
return 0;
}
else {
DBG_ERROR(NULL, "Not of type AQH_NODE_SERVER object");
return GWEN_ERROR_INVALID;
}
}
void _readConfig(AQH_OBJECT *o, AQH_NODE_SERVER *xo, GWEN_DB_NODE *dbArgs)
{
xo->dbArgs=dbArgs;
xo->timeout=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 0);
xo->noAttn=GWEN_DB_GetIntValue(dbArgs, "noAttn", 0, 0);
xo->nodeAddress=GWEN_DB_GetIntValue(dbArgs, "nodeAddress", 0, AQHOMED_DEFAULT_NODEADDR);
AQH_NodeServer_SetDbFile(o, GWEN_DB_GetCharValue(dbArgs, "dbfile", 0, NULL));
AQH_NodeServer_SetLogFile(o, GWEN_DB_GetCharValue(dbArgs, "logfile", 0, NULL));
AQH_NodeServer_SetDevicePath(o, GWEN_DB_GetCharValue(dbArgs, "device", 0, AQHOMED_DEFAULT_DEVICE));
AQH_NodeServer_SetTpcAddress(o, readCharConfigWithAlt(dbArgs, "tcpAddress", "ConfigFile/nodesAddress", NULL));
AQH_NodeServer_SetTcpPort(o, readIntConfigWithAlt(dbArgs, "tcpPort", "ConfigFile/nodesPort", AQHOMED_DEFAULT_IPC_PORT, -1));
AQH_NodeServer_SetBrokerAddress(o, readCharConfigWithAlt(dbArgs, "brokerAddress", "ConfigFile/brokerAddress", "127.0.0.1"));
AQH_NodeServer_SetBrokerPort(o, readIntConfigWithAlt(dbArgs, "brokerPort", "ConfigFile/brokerPort", AQHOMED_DEFAULT_BROKER_PORT, -1));
AQH_NodeServer_SetBrokerClientId(o, GWEN_DB_GetCharValue(dbArgs, "brokerClientId", 0, AQHOMED_DEFAULT_BROKER_CLIENTID));
}
const char *readCharConfigWithAlt(GWEN_DB_NODE *dbArgs, const char *varName, const char *altVarName, const char *defaultValue)
{
const char *s;
s=GWEN_DB_GetCharValue(dbArgs, varName, 0, NULL);
if (!(s && *s))
s=GWEN_DB_GetCharValue(dbArgs, altVarName, 0, NULL);
return (s && *s)?s:defaultValue;
}
int readIntConfigWithAlt(GWEN_DB_NODE *dbArgs, const char *varName, const char *altVarName, int defaultValue, int nonValue)
{
int i;
i=GWEN_DB_GetIntValue(dbArgs, varName, 0, nonValue);
if (i==nonValue)
i=GWEN_DB_GetIntValue(dbArgs, altVarName, 0, nonValue);
return (i!=nonValue)?i:defaultValue;
}
int _startIpc(AQH_OBJECT *o, AQH_NODE_SERVER *xo)
{
if (xo->ipcEndpoint) {
AQH_Object_Disable(xo->ipcEndpoint);
AQH_Object_free(xo->ipcEndpoint);
xo->ipcEndpoint=NULL;
}
if (xo->tcpAddress && *(xo->tcpAddress) && xo->tcpPort>0) {
int fd;
DBG_ERROR(NULL, "Starting IPC service on \"%s\":%d", xo->tcpAddress, xo->tcpPort);
fd=AQH_TcpdObject_CreateListeningSocket(xo->tcpAddress, xo->tcpPort);
if (fd<0) {
DBG_INFO(NULL, "here");
return GWEN_ERROR_IO;
}
xo->ipcEndpoint=AQH_IpcServerObject_new(AQH_Object_GetEventLoop(o), fd);
AQH_Object_AddLink(xo->ipcEndpoint, AQH_IPC_SERVER_SIGNAL_NEWCLIENT, AQH_NODE_SERVER_SLOT_NEWCLIENT, o);
AQH_Object_Enable(xo->ipcEndpoint);
return 0;
}
else {
DBG_ERROR(NULL, "Missing server address");
return GWEN_ERROR_GENERIC;
}
}
int _startTty(AQH_OBJECT *o, AQH_NODE_SERVER *xo)
{
if (xo->ttyEndpoint) {
AQH_Object_Disable(xo->ttyEndpoint);
AQH_Object_free(xo->ttyEndpoint);
xo->ttyEndpoint=NULL;
}
if (xo->devicePath && *(xo->devicePath)) {
int fd;
DBG_ERROR(NULL, "Opening TTY device \"%s\"", xo->devicePath);
fd=AQH_TtyObject_OpenAndInitDevice(xo->devicePath, &(xo->initialTermiosState));
if (fd<0) {
DBG_INFO(NULL, "here");
return GWEN_ERROR_IO;
}
xo->ttyEndpoint=AQH_TtyEndpoint2_new(AQH_Object_GetEventLoop(o), fd, xo->noAttn);
AQH_Object_AddLink(xo->ttyEndpoint, AQH_ENDPOINT_SIGNAL_CLOSED, AQH_NODE_SERVER_SLOT_TTYCLOSED, o);
AQH_Object_Enable(xo->ttyEndpoint);
}
else {
DBG_ERROR(NULL, "Missing device path");
return GWEN_ERROR_GENERIC;
}
return 0;
}
int _startBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo)
{
if (xo->brokerEndpoint) {
AQH_Object_Disable(xo->brokerEndpoint);
AQH_Object_free(xo->brokerEndpoint);
xo->brokerEndpoint=NULL;
}
if (xo->brokerAddress && *(xo->brokerAddress) && xo->brokerPort) {
AQH_OBJECT *ep;
int fd;
int rv;
fd=AQH_TcpObject_CreateConnectedSocket(xo->brokerAddress, xo->brokerPort);
if (fd<0) {
DBG_ERROR(NULL, "Error connecting to broker server %s:%d", xo->brokerAddress, xo->brokerPort);
return GWEN_ERROR_IO;
}
DBG_ERROR(NULL, "Physically connected to broker server %s:%d", xo->brokerAddress, xo->brokerPort);
ep=AQH_IpcClientObject_new(AQH_Object_GetEventLoop(o), fd);
assert(ep);
AQH_Endpoint_SetServiceName(ep, xo->brokerClientId);
AQH_Object_AddLink(ep, AQH_ENDPOINT_SIGNAL_CLOSED, AQH_NODE_SERVER_SLOT_BROKERCLOSED, o);
AQH_Object_Enable(ep);
xo->brokerEndpoint=ep;
rv=_exchangeConnect(xo, 0);
if (rv!=0) {
DBG_ERROR(NULL, "Error connecting to broker: %d", rv);
return (rv<0)?rv:GWEN_ERROR_PERMISSIONS;
}
DBG_ERROR(NULL, "Fully connected to broker at %s:%d", xo->brokerAddress, xo->brokerPort);
return 0;
}
else {
DBG_ERROR(NULL, "No server settings");
return GWEN_ERROR_BAD_DATA;
}
return 0;
}
int _exchangeConnect(AQH_NODE_SERVER *xo, uint32_t flags)
{
AQH_MESSAGE *msgOut;
uint32_t msgId;
msgId=AQH_Endpoint_GetNextMessageId(xo->brokerEndpoint);
msgOut=AQH_IpcMessageConnect_new(AQH_IPC_PROTOCOL_DATA_ID, AQH_IPC_PROTOCOL_DATA_VERSION,
AQH_MSGTYPE_IPC_CONNECT_REQ,
msgId, 0,
xo->brokerClientId, NULL, NULL, flags);
AQH_Endpoint_AddMsgOut(xo->brokerEndpoint, msgOut);
return AQH_IpcEndpoint_WaitForResultMsg(xo->brokerEndpoint,
AQH_IPC_PROTOCOL_DATA_ID, AQH_IPC_PROTOCOL_DATA_VERSION, AQH_MSGTYPE_IPC_RESULT,
msgId, xo->timeoutInSeconds);
}
void _setupDb(AQH_NODE_SERVER *xo)
{
if (xo->dbFile) {
GWEN_DB_NODE *dbNodeDb;
int rv;
if (xo->nodeDb)
AQH_NodeDb_free(xo->nodeDb);
xo->nodeDb=AQH_NodeDb_new();
dbNodeDb=GWEN_DB_Group_new("dbNodes");
rv=GWEN_DB_ReadFile(dbNodeDb, xo->dbFile, GWEN_DB_FLAGS_DEFAULT|GWEN_PATH_FLAGS_CREATE_GROUP);
if (rv==0) {
AQH_NodeDb_fromDb(xo->nodeDb, dbNodeDb);
}
GWEN_DB_Group_free(dbNodeDb);
}
}
int _loadDeviceList(AQH_NODE_SERVER *xo)
{
AQHNODE_DEVICE_LIST *deviceList;
deviceList=AQH_NodeServer_ReadDataDeviceFiles();
if (deviceList==NULL) {
DBG_ERROR(NULL, "Error reading device list");
return GWEN_ERROR_GENERIC;
}
xo->deviceDefList=deviceList;
return 0;
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* fini
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
void AQH_NodeServer_Fini(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
AQH_OBJECT *ep;
ep=AQH_Object_List_First(xo->ipcClientList);
while(ep) {
AQH_OBJECT *epNext;
epNext=AQH_Object_List_Next(ep);
AQH_Object_List_Del(ep);
AQH_Object_free(ep);
ep=epNext;
} /* while */
AQH_Object_free(xo->brokerEndpoint);
xo->brokerEndpoint=NULL;
AQH_Object_free(xo->ttyEndpoint);
xo->ttyEndpoint=NULL;
AQH_NodeServer_WriteNodeDb(o);
if (xo->pidFile)
remove(xo->pidFile);
}
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* client management functions
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
void AQH_NodeServer_CleanupClients(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
AQH_OBJECT *ep;
ep=AQH_Object_List_First(xo->ipcClientList);
while(ep) {
AQH_OBJECT *epNext;
epNext=AQH_Object_List_Next(ep);
if (AQH_Object_GetFlags(ep) & AQH_OBJECT_FLAGS_DELETE) {
AQH_Object_List_Del(ep);
AQH_Object_free(ep);
}
ep=epNext;
} /* while */
}
}
void AQH_NodeServer_HandleClientMsgs(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
AQH_OBJECT *ep;
ep=AQH_Object_List_First(xo->ipcClientList);
while(ep) {
AQH_OBJECT *epNext;
epNext=AQH_Object_List_Next(ep);
_handleMsgsFromClient(o, xo, ep);
ep=epNext;
} /* while */
}
}
void _handleMsgsFromClient(AQH_OBJECT *o, AQH_NODE_SERVER *xo, AQH_OBJECT *ep)
{
AQH_MESSAGE *msg;
while( (msg=AQH_Endpoint_GetNextMsgIn(ep)) ) {
AQH_Message_SetObject(msg, ep);
if (AQH_Request_Tree2_HandleIpcMsg(xo->requestTree, ep, msg)!=AQH_MSG_REQUEST_RESULT_HANDLED)
_handleMsgFromClient(o, xo, ep, msg);
AQH_Message_free(msg);
}
}
void _handleMsgFromClient(GWEN_UNUSED AQH_OBJECT *o, GWEN_UNUSED AQH_NODE_SERVER *xo, GWEN_UNUSED AQH_OBJECT *ep, const AQH_MESSAGE *msg)
{
uint16_t code;
uint8_t protoId;
/* exec IPC message */
code=AQH_IpcMessage_GetCode(msg);
protoId=AQH_IpcMessage_GetProtoId(msg);
if (protoId==AQH_IPC_PROTOCOL_NODES_ID) {
GWEN_TAG16_LIST *tagList;
tagList=AQH_IpcMessageTag16_ParsePayload(msg, 0);
DBG_ERROR(NULL, "Received IPC packet %d (%x)", (int) code, code);
switch(code) {
case AQH_MSGTYPE_IPC_NODES_CONNECT_REQ: AQH_NodeServer_HandleConnect(o, ep, msg, tagList); break;
case AQH_MSGTYPE_IPC_NODES_FORWARD: AQH_NodeServer_HandleForward(o, ep, msg, tagList); break;
case AQH_MSGTYPE_IPC_NODES_SETACCMSGGRPS: AQH_NodeServer_HandleSetAccMsgGrps(o, ep, msg, tagList); break;
case AQH_MSGTYPE_IPC_NODES_GETDEVICES_REQ: AQH_NodeServer_HandleGetNodes(o, ep, msg); break;
default: break;
}
GWEN_Tag16_List_free(tagList);
}
else {
DBG_ERROR(NULL, "Invalid IPC protocol %d (%02x)", protoId, protoId);
}
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* TTY management functions
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
void AQH_NodeServer_HandleTtyMsgs(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo && xo->ttyEndpoint) {
AQH_MESSAGE *msg;
while( (msg=AQH_Endpoint_GetNextMsgIn(xo->ttyEndpoint)) ) {
_writeTtyMsgToLogFile(xo, msg);
AQH_Message_SetObject(msg, xo->ttyEndpoint);
if (AQH_Request_Tree2_HandleTtyMsg(xo->requestTree, xo->ttyEndpoint, msg)!=AQH_MSG_REQUEST_RESULT_HANDLED)
_handleMsgFromTty(o, xo, msg);
AQH_Message_free(msg);
}
}
}
void _handleMsgFromTty(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint8_t code;
code=AQH_NodeMessage_GetMsgType(msg);
DBG_INFO(NULL, "Received Node packet %d (%x)", (int) code, code);
AQH_NodeServer_NodeMsgToDb(o, msg);
_forwardTtyMsgToBroker(o, xo, msg);
_forwardTtyMsgToClients(xo, msg);
}
void _forwardTtyMsgToBroker(AQH_OBJECT *o, AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint8_t code;
code=AQH_NodeMessage_GetMsgType(msg);
switch(code) {
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 _forwardTtyMsgToClients(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;
DBG_ERROR(NULL, "Forwarding node message %d to client", AQH_NodeMessage_GetMsgType(msg));
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;
uint32_t uid;
AQH_NODE_INFO *ni;
const char *devName;
const AQHNODE_DEVICE *devInfo;
const AQHNODE_VALUE *value;
const char *vname;
valueId=AQH_ValueMessage_GetValueId(msg);
v=AQH_ValueMessage_GetValue(msg);
uid=AQH_ValueMessage_GetUid(msg);
ni=uid?AQH_NodeDb_GetNodeInfoByUid(xo->nodeDb, uid):NULL;;
devName=ni?AQH_NodeInfo_GetDeviceId(ni):NULL;
devInfo=devName?AQH_NodeServer_GetDeviceDefByName(o, devName):NULL;
value=devInfo?AQHNODE_Value_List_GetById(AQHNODE_Device_GetValueList(devInfo), valueId):NULL;
vname=value?AQHNODE_Value_GetName(value):NULL;
if (vname && *vname)
_publishDouble(xo, uid, vname, AQHNODE_Value_GetModality(value), AQHNODE_Value_GetValueUnits(value), v);
}
void _forwardDataFromSendStatsMsgToBroker(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint16_t packetsOutInt;
packetsOutInt=AQH_SendStatsMessage_GetPacketsOut(msg);
if (packetsOutInt) {
uint32_t uid;
double packetsOut;
double collisions;
double busy;
double collisionsPercentage=0.0;
double busyPercentage=0.0;
uid=AQH_SendStatsMessage_GetUid(msg);
packetsOut=/*(double)*/ packetsOutInt;
collisions=/*(double)*/ AQH_SendStatsMessage_GetCollisions(msg);
busy=/*(double)*/ AQH_SendStatsMessage_GetBusyErrors(msg);
collisionsPercentage=collisions*100.0/packetsOut;
busyPercentage=busy*100.0/packetsOut;
_publishInt( xo, uid, "net/packetsOut", 0, NULL, packetsOutInt);
_publishInt( xo, uid, "net/collisions", 0, NULL, (int) AQH_SendStatsMessage_GetCollisions(msg));
_publishDouble(xo, uid, "net/collisionsPercent", 0, "%", collisionsPercentage);
_publishDouble(xo, uid, "net/busyPercent", 0, "%", busyPercentage);
}
}
void _forwardDataFromRecvStatsMsgToBroker(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
uint16_t packetsInInt;
packetsInInt=AQH_RecvStatsMessage_GetPacketsIn(msg);
if (packetsInInt) {
uint32_t uid;
double packetsIn;
double crcErrors;
double ioErrors;
double crcErrorsPercentage=0.0;
double ioErrorsPercentage=0.0;
uid=AQH_SendStatsMessage_GetUid(msg);
packetsIn=/*(double)*/ packetsInInt;
crcErrors=/*(double)*/AQH_RecvStatsMessage_GetCrcErrors(msg);
ioErrors=/*(double)*/AQH_RecvStatsMessage_GetIoErrors(msg);
crcErrorsPercentage=crcErrors*100.0/packetsIn;
ioErrorsPercentage=ioErrors*100.0/packetsIn;
_publishInt( xo, uid, "net/packetsIn", 0, NULL, packetsInInt);
_publishInt( xo, uid, "net/crcerrors", 0, NULL, (int) AQH_RecvStatsMessage_GetCrcErrors(msg));
_publishInt( xo, uid, "net/ioerrors", 0, NULL, (int) AQH_RecvStatsMessage_GetIoErrors(msg));
_publishDouble(xo, uid, "net/crcerrorsPercent", 0, "%", crcErrorsPercentage);
_publishDouble(xo, uid, "net/ioerrorsPercent", 0, "%", ioErrorsPercentage);
}
}
void _publishInt(AQH_NODE_SERVER *xo, uint32_t uid, const char *vPath, int vModality, const char *vUnits, int v)
{
_publishDouble(xo, uid, vPath, vModality, vUnits, /*(double)*/ v);
}
void _publishDouble(AQH_NODE_SERVER *xo, uint32_t uid, const char *vPath, int vModality, const char *vUnits, double v)
{
union {double f; uint64_t i;} u;
uint64_t arrayToSend[2];
AQH_VALUE *value;
u.f=v;
arrayToSend[0]=(uint64_t) time(NULL);
arrayToSend[1]=u.i;
value=AQH_Value_new();
_setDeviceName(value, uid);
AQH_Value_SetName(value, vPath);
AQH_Value_SetValueUnits(value, vUnits);
AQH_Value_SetValueType(value, AQH_ValueType_Sensor);
AQH_Value_SetModality(value, vModality);
if (xo->brokerEndpoint) {
AQH_MESSAGE *pubMsg;
pubMsg=AQH_IpcdMessageMultiData_new(AQH_MSGTYPE_IPC_DATA_UPDATEDATA,
AQH_Endpoint_GetNextMessageId(xo->brokerEndpoint), 0,
value, arrayToSend, 1);
if (pubMsg) {
DBG_INFO(AQH_LOGDOMAIN,
"BROKER PUBLISH %s(%s/%s): %f",
AQH_Value_GetName(value),
AQH_Value_GetDeviceName(value),
AQH_Value_GetName(value),
v);
AQH_Endpoint_AddMsgOut(xo->brokerEndpoint, pubMsg);
}
}
else {
DBG_ERROR(AQH_LOGDOMAIN,
"Skipping BROKER PUBLISH %s(%s/%s): %f (no broker connection)",
AQH_Value_GetName(value),
AQH_Value_GetDeviceName(value),
AQH_Value_GetName(value),
v);
}
AQH_Value_free(value);
}
void _setDeviceName(AQH_VALUE *value, uint32_t uid)
{
GWEN_BUFFER *buf;
buf=GWEN_Buffer_new(0, 64, 0, 1);
GWEN_Buffer_AppendArgs(buf, "%08x", uid);
AQH_Value_SetDeviceName(value, GWEN_Buffer_GetStart(buf));
GWEN_Buffer_free(buf);
}
void _writeTtyMsgToLogFile(AQH_NODE_SERVER *xo, const AQH_MESSAGE *msg)
{
if (xo->logFile) {
GWEN_BUFFER *dbuf;
GWEN_TIME *ti;
dbuf=GWEN_Buffer_new(0, 256, 0, 1);
ti=GWEN_CurrentTime();
GWEN_Time_toString(ti, "YYYY-MM-DD hh:mm:ss ", dbuf);
GWEN_Time_free(ti);
ti=NULL;
AQH_NodeMessage_DumpSpecificToBuffer(msg, dbuf, "received");
_writeToLogFile(xo->logFile, GWEN_Buffer_GetStart(dbuf));
GWEN_Buffer_free(dbuf);
}
}
void _writeToLogFile(const char *filename, const char *txt)
{
if (txt && *txt) {
FILE *f;
f=fopen(filename, "a+");
if (f) {
if (1!=fwrite(txt, strlen(txt), 1, f)) {
DBG_ERROR(NULL, "Error logging.");
}
fclose(f);
}
}
}
void AQH_NodeServer_CheckTtyConnection(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
if (xo->ttyEndpoint) {
if (AQH_Object_GetFlags(xo->ttyEndpoint) & AQH_OBJECT_FLAGS_DELETE) {
DBG_ERROR(NULL, "Deleting TTY connection");
AQH_Object_Disable(xo->ttyEndpoint);
AQH_Object_free(xo->ttyEndpoint);
xo->ttyEndpoint=NULL;
}
}
if (xo->ttyEndpoint==NULL) {
time_t now;
now=time(NULL);
if (_diffInSeconds(now, xo->timestampBrokerDown)>AQH_NODE_SERVER_TTY_RESTARTTIME) {
int rv;
DBG_ERROR(NULL, "Re-opening TTY device");
rv=_startTty(o, xo);
if (rv<0) {
DBG_ERROR(NULL, "here (%d)", rv);
}
}
}
}
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* broker management functions
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
void AQH_NodeServer_HandleBrokerMsgs(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo && xo->brokerEndpoint) {
AQH_MESSAGE *msg;
while( (msg=AQH_Endpoint_GetNextMsgIn(xo->brokerEndpoint)) ) {
AQH_Message_SetObject(msg, xo->brokerEndpoint);
if (AQH_Request_Tree2_HandleIpcMsg(xo->requestTree, xo->brokerEndpoint, msg)!=AQH_MSG_REQUEST_RESULT_HANDLED)
_handleMsgFromBroker(o, xo->brokerEndpoint, msg);
AQH_Message_free(msg);
}
}
}
void _handleMsgFromBroker(AQH_OBJECT *o, AQH_OBJECT *ep, const AQH_MESSAGE *msg)
{
uint16_t code;
uint8_t protoId;
/* exec IPC message */
code=AQH_IpcMessage_GetCode(msg);
protoId=AQH_IpcMessage_GetProtoId(msg);
if (protoId==AQH_IPC_PROTOCOL_DATA_ID) {
DBG_INFO(NULL, "Received IPC packet %d (%x)", (int) code, code);
switch(code) {
case AQH_MSGTYPE_IPC_DATA_SETDATA: AQH_NodeServer_HandleSetData(o, ep, msg); break;
default: break;
}
}
else {
DBG_ERROR(NULL, "Invalid IPC protocol %d (%02x)", protoId, protoId);
}
}
void AQH_NodeServer_CheckBrokerConnection(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo && xo->dbArgs) {
if (xo->brokerEndpoint) {
if (AQH_Object_GetFlags(xo->brokerEndpoint) & AQH_OBJECT_FLAGS_DELETE) {
DBG_ERROR(NULL, "Deleting broker connection");
AQH_Object_Disable(xo->brokerEndpoint);
AQH_Object_free(xo->brokerEndpoint);
xo->brokerEndpoint=NULL;
}
}
if (xo->brokerEndpoint==NULL) {
time_t now;
now=time(NULL);
if (_diffInSeconds(now, xo->timestampBrokerDown)>AQH_NODE_SERVER_BROKER_RESTARTTIME) {
int rv;
DBG_ERROR(NULL, "Restarting broker connection");
rv=_startBroker(o, xo);
if (rv<0) {
DBG_ERROR(NULL, "here (%d)", rv);
}
}
}
}
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* request management functions
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
AQH_MSG_REQUEST *AQH_NodeServer_GetRequestTree(const AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo)
return xo->requestTree;
return NULL;
}
void AQH_NodeServer_AddRequestToTree(AQH_OBJECT *o, AQH_MSG_REQUEST *rq)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo && rq)
AQH_MsgRequest_Tree2_AddChild(xo->requestTree, rq);
}
void AQH_NodeServer_CleanupRequests(AQH_OBJECT *o)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
AQH_Request_Tree2_CheckTimeouts(xo->requestTree);
AQH_Request_Tree2_Cleanup(xo->requestTree);
}
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* signal handler
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
int _handleSignal(AQH_OBJECT *o, uint32_t slotId, AQH_OBJECT *senderObject, GWEN_UNUSED int param1, void *param2)
{
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo) {
switch(slotId) {
case AQH_NODE_SERVER_SLOT_NEWCLIENT: return _handleNewIpcClient(o, xo, (AQH_OBJECT*) param2);
case AQH_NODE_SERVER_SLOT_CLIENTCLOSED: return _handleIpcClientDown(senderObject);
case AQH_NODE_SERVER_SLOT_BROKERCLOSED: return _handleBrokerDown(xo);
case AQH_NODE_SERVER_SLOT_TTYCLOSED: return _handleTtyDown(xo);
default:
break;
}
}
return 0; /* not handled */
}
int _handleNewIpcClient(AQH_OBJECT *o, AQH_NODE_SERVER *xo, AQH_OBJECT *clientEndpoint)
{
DBG_ERROR(NULL, "New IPC client");
AQH_Object_AddLink(clientEndpoint, AQH_ENDPOINT_SIGNAL_CLOSED, AQH_NODE_SERVER_SLOT_CLIENTCLOSED, o);
AQH_Object_List_Add(clientEndpoint, xo->ipcClientList);
return 1; /* handled */
}
int _handleIpcClientDown(AQH_OBJECT *clientEndpoint)
{
DBG_ERROR(NULL, "IPC client down");
AQH_Object_AddFlags(clientEndpoint, AQH_OBJECT_FLAGS_DELETE);
return 1; /* handled */
}
int _handleBrokerDown(AQH_NODE_SERVER *xo)
{
if (xo->brokerEndpoint) {
DBG_ERROR(NULL, "Broker connection down");
AQH_Object_AddFlags(xo->brokerEndpoint, AQH_OBJECT_FLAGS_DELETE);
xo->timestampBrokerDown=time(NULL);
}
return 1;
}
int _handleTtyDown(AQH_NODE_SERVER *xo)
{
if (xo->ttyEndpoint) {
DBG_ERROR(NULL, "TTY closed");
AQH_Object_AddFlags(xo->ttyEndpoint, AQH_OBJECT_FLAGS_DELETE);
xo->timestampTtyDown=time(NULL);
}
return 1;
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* device management functions
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
const AQHNODE_DEVICE_LIST *AQH_NodeServer_GetDeviceDefList(const AQH_OBJECT *o)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo)
return xo->deviceDefList;
}
return NULL;
}
const AQHNODE_DEVICE *AQH_NodeServer_FindDeviceDef(const AQH_OBJECT *o, uint32_t manufacturer, uint16_t deviceType, uint16_t deviceVersion)
{
if (o) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo && xo->deviceDefList) {
const AQHNODE_DEVICE *device;
device=AQHNODE_Device_List_First(xo->deviceDefList);
while(device) {
if (AQHNODE_Device_GetManufacturer(device)==manufacturer &&
AQHNODE_Device_GetDeviceType(device)==deviceType &&
AQHNODE_Device_GetDeviceVersion(device)==(deviceVersion & 0xff00))
return device;
device=AQHNODE_Device_List_Next(device);
}
}
}
return NULL;
}
const AQHNODE_DEVICE *AQH_NodeServer_GetDeviceDefByName(const AQH_OBJECT *o, const char *name)
{
if (o && name) {
AQH_NODE_SERVER *xo;
xo=GWEN_INHERIT_GETDATA(AQH_OBJECT, AQH_NODE_SERVER, o);
if (xo && xo->deviceDefList)
return AQHNODE_Device_List_GetByName(xo->deviceDefList, name);
}
return NULL;
}
/* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
* helper functions
* xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
*/
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 _diffInSeconds(time_t t1, time_t t0)
{
return t1-t0;
}
int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs)
{
int rv;
const GWEN_ARGS args[]= {
/* flags type name min max s long short_descr, long_descr */
{ A_ARG, A_CHAR, "loglevel", 0, 1, "L", "loglevel", I18S("Specify loglevel"), NULL},
{ A_ARG, A_CHAR, "cfgdir", 0, 1, "D", "cfgdir", I18S("Specify the configuration folder"), NULL},
{ A_ARG, A_CHAR, "charset", 0, 1, NULL, "charset", I18S("Specify the output character set"), NULL},
{ A_ARG, A_CHAR, "device", 0, 1, "d", "device", I18S("Specify the device (e.g. /dev/ttyUSB0)"), NULL},
{ A_ARG, A_INT, "nodeAddress", 0, 1, "n", "node", I18S("AqHome node adaptor address(default 240)"), NULL},
{ A_ARG, A_CHAR, "logFile", 0, 1, "l", "logfile", I18S("Specify a logfile to log received messages to"), NULL},
{ A_ARG, A_CHAR, "tcpAddress", 0, 1, "t", "tcpaddress", I18S("TCP address to listen on (disabled if missing)"), NULL},
{ A_ARG, A_INT, "tcpPort", 0, 1, "P", "tcpport", I18S("TCP port to listen on (default: 45454)"), NULL},
{ A_ARG, A_CHAR, "brokerAddress", 0, 1, "ba", "brokerddress", I18S("Broker address [127.0.0.1]"), NULL},
{ A_ARG, A_INT, "brokerPort", 0, 1, "bp", "brokerport", I18S("Broker port [1899]"), NULL},
{ A_ARG, A_CHAR, "brokerClientId", 0, 1, NULL, "brokerclientid", I18S("Broker client id"), NULL},
{ A_ARG, A_CHAR, "dbfile", 0, 1, "db", "dbfile", I18S("DB file to read/write node database"), NULL},
{ A_ARG, A_CHAR, "pidfile", 0, 1, "p", "pidfile", I18S("PID file"), NULL},
{ A_ARG, A_INT, "timeout", 0, 1, "T", NULL, I18S("timeout in seconds [0]"), NULL},
{ 0, A_INT, "noAttn", 0, 1, "N", "noattn", I18S("Don't use ATTN line (for T03 or newer)"), NULL},
{ A_END, A_INT, "help", 0, 0, "h", "help", I18S("Show this help screen"), NULL}
};
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;
}
Reference in New Issue View Git Blame Copy Permalink