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

More work on mqtt tool.

Browse Source
This commit is contained in:
Martin Preuss
2023-10-04 16:02:02 +02:00
parent 4730943931
commit bfed937950
27 changed files with 1527 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files

314
apps/aqhome-mqttlog/main.c
View File

@@ -11,12 +11,12 @@
#endif
#include "./item.h"
#include "./mqtt.h"
#include "./messages.h"
#include "./init.h"
#include "./fini.h"
#include "./loop.h"
#include "./loop_mqtt.h"
#include "aqhome/aqhome.h"
#include "aqhome/mqtt/msg_mqtt.h"
#include "aqhome/mqtt/msg_mqtt_publish.h"
#include <gwenhywfar/gwenhywfar.h>
#include <gwenhywfar/args.h>
@@ -49,7 +49,7 @@
#define AQHOME_MQTTLOG_PING_INTERVAL 120
//#define FULL_DEBUG
#define FULL_DEBUG
@@ -58,9 +58,7 @@
* ------------------------------------------------------------------------------------------------
*/
static int _serve(GWEN_DB_NODE *dbArgs);
static int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs);
static int _createPidFile(const char *pidFilename);
static int _serve(AQHOME_MQTT *aqh);
#ifdef HAVE_SIGNAL_H
static int _setSignalHandlers(void);
@@ -87,6 +85,7 @@ static int stopService=0;
int main(int argc, char **argv)
{
AQHOME_MQTT *aqh;
GWEN_DB_NODE *dbArgs;
int rv;
GWEN_GUI *gui;
@@ -108,16 +107,16 @@ int main(int argc, char **argv)
return 2;
}
dbArgs=GWEN_DB_Group_new("arguments");
rv=_readArgs(argc, argv, dbArgs);
aqh=AqHomeMqtt_new();
rv=AqHomeMqtt_Init(aqh, argc, argv);
if (rv<0) {
if (rv==GWEN_ERROR_CLOSE)
return 1;
DBG_INFO(NULL, "here (%d)", rv);
return 2;
}
else if (rv==1) {
DBG_INFO(NULL, "Help printed, done");
return 0;
}
dbArgs=AqHomeMqtt_GetDbArgs(aqh);
gui=GWEN_Gui_CGui_new();
s=GWEN_DB_GetCharValue(dbArgs, "charset", 0, NULL);
@@ -125,13 +124,15 @@ int main(int argc, char **argv)
GWEN_Gui_SetCharSet(gui, s);
GWEN_Gui_SetGui(gui);
rv=_serve(dbArgs);
rv=_serve(aqh);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return 2;
}
GWEN_DB_Group_free(dbArgs);
AqHomeMqtt_Fini(aqh);
AqHomeMqtt_free(aqh);
GWEN_Gui_SetGui(NULL);
GWEN_Gui_free(gui);
@@ -140,24 +141,18 @@ int main(int argc, char **argv)
int _serve(GWEN_DB_NODE *dbArgs)
int _serve(AQHOME_MQTT *aqh)
{
const char *pidFile;
GWEN_MSG_ENDPOINT *epTcp;
ITEM_LIST *itemList;
int rv;
int timeout;
time_t startTime;
time_t lastPingSendTime;
const char *baseFolder;
GWEN_DB_NODE *dbArgs;
startTime=time(NULL);
itemList=AqHomeMqttLog_ReadItems(dbArgs);
if (itemList==NULL) {
DBG_ERROR(NULL, "No items to listen to, aborting.");
return GWEN_ERROR_GENERIC;
}
dbArgs=AqHomeMqtt_GetDbArgs(aqh);
rv=_setSignalHandlers();
if (rv<0) {
@@ -165,80 +160,15 @@ int _serve(GWEN_DB_NODE *dbArgs)
return rv;
}
baseFolder=GWEN_DB_GetCharValue(dbArgs, "writeToFolder", 0, "/tmp/aqhome");
timeout=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 0);
pidFile=GWEN_DB_GetCharValue(dbArgs, "pidfile", 0, "aqhome-mqttlog.pid");
if (pidFile && *pidFile) {
rv=_createPidFile(pidFile);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
return rv;
}
}
epTcp=AqHomeMqttLog_CreateMqttEndpoint(dbArgs);
if (epTcp==NULL) {
DBG_INFO(NULL, "here");
Item_List_free(itemList);
return GWEN_ERROR_GENERIC;
}
rv=AqHomeMqttLog_MqttConnect(epTcp);
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
GWEN_MsgEndpoint_free(epTcp);
Item_List_free(itemList);
return rv;
}
rv=AqHomeMqttLog_Subscribe(epTcp, "#");
if (rv<0) {
DBG_INFO(NULL, "here (%d)", rv);
GWEN_MsgEndpoint_free(epTcp);
Item_List_free(itemList);
return rv;
}
lastPingSendTime=time(NULL);
epTcp=AqHomeMqtt_GetMqttEndpoint(aqh);
while(!stopService) {
DBG_DEBUG(NULL, "Next loop");
GWEN_MsgEndpoint_IoLoop(epTcp, 2000); /* 2000 ms */
if (GWEN_MsgEndpoint_GetState(epTcp)!=GWEN_MSG_ENDPOINT_STATE_CONNECTED) {
DBG_INFO(NULL, "Not connected...");
}
else {
GWEN_MSG *msg;
while( (msg=GWEN_MsgEndpoint_TakeFirstReceivedMessage(epTcp)) ) {
#ifdef FULL_DEBUG
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
#endif
if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(AQH_MQTTMSG_MSGTYPE_PUBLISH & 0xf0)) {
#ifdef FULL_DEBUG
GWEN_BUFFER *buf;
buf=GWEN_Buffer_new(0, 256, 0, 1);
AQH_PublishMqttMsg_DumpToBuffer(msg, buf, "received");
fprintf(stdout, "%s\n", GWEN_Buffer_GetStart(buf));
GWEN_Buffer_free(buf);
#endif
AqHomeMqttLog_HandlePublishMsg(baseFolder, itemList, msg);
}
else if ((AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0)==(AQH_MQTTMSG_MSGTYPE_PINGRESP & 0xf0)) {
DBG_INFO(AQH_LOGDOMAIN, "PING response received");
}
else {
#ifdef FULL_DEBUG
DBG_ERROR(NULL, "Received this message:");
GWEN_Text_DumpString((const char*) GWEN_Msg_GetConstBuffer(msg), GWEN_Msg_GetBytesInBuffer(msg), 2);
#endif
}
GWEN_Msg_free(msg);
}
}
AqHomeMqttLog_Loop(aqh, 2000);
if (timeout) {
time_t now;
@@ -254,7 +184,7 @@ int _serve(GWEN_DB_NODE *dbArgs)
now=time(NULL);
if (now-lastPingSendTime>AQHOME_MQTTLOG_PING_INTERVAL) {
rv=AqHomeMqttLog_Ping(epTcp);
rv=AqHomeMqttLog_SendPing(aqh);
if (rv<0) {
DBG_INFO(NULL, "Error sending PING");
}
@@ -263,11 +193,7 @@ int _serve(GWEN_DB_NODE *dbArgs)
}
}
if (pidFile && *pidFile)
remove(pidFile);
GWEN_MsgEndpoint_free(epTcp);
Item_List_free(itemList);
return 0;
}
@@ -339,195 +265,3 @@ void _signalHandler(int s)
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 _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs)
{
int rv;
const GWEN_ARGS args[]= {
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"cfgdir", /* name */
0, /* minnum */
1, /* maxnum */
"D", /* short option */
"cfgdir", /* long option */
I18S("Specify the configuration folder"),
I18S("Specify the configuration folder")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"charset", /* name */
0, /* minnum */
1, /* maxnum */
0, /* short option */
"charset", /* long option */
I18S("Specify the output character set"), /* short description */
I18S("Specify the output character set") /* long description */
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"mqttAddress", /* name */
0, /* minnum */
1, /* maxnum */
"ma", /* short option */
"mqttaddress", /* long option */
I18S("Specify the address of the MQTT server to connect to (disabled if missing)"),
I18S("Specify the address of the MQTT server to connect to (disabled if missing)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"mqttPort", /* name */
0, /* minnum */
1, /* maxnum */
"mp", /* short option */
"mqttport", /* long option */
I18S("Specify the port of the MQTT server (default: 1883)"),
I18S("Specify the port of the MQTT server (default: 1883)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"mqttClientId", /* name */
0, /* minnum */
1, /* maxnum */
NULL, /* short option */
"mqttclientid", /* long option */
I18S("Specify client id for the MQTT server (default: \"aqhomed\")"),
I18S("Specify client id for the MQTT server (default: \"aqhomed\")")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"mqttKeepAlive", /* name */
0, /* minnum */
1, /* maxnum */
"mk", /* short option */
"mqttkeepalive", /* long option */
I18S("Specify keepalive time in seconds (defaults: 600)"),
I18S("Specify keepalive time in seconds (defaults: 600)")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"writeToFolder", /* name */
0, /* minnum */
1, /* maxnum */
"W", /* short option */
NULL, /* long option */
I18S("Specify folder to write received values to"),
I18S("Specify folder to write received values to")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"pidfile", /* name */
0, /* minnum */
1, /* maxnum */
"p", /* short option */
"pidfile", /* long option */
I18S("Specify the PID file"),
I18S("Specify the PID file")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Char, /* type */
"itemfile", /* name */
0, /* minnum */
1, /* maxnum */
"i", /* short option */
"itemfile", /* long option */
I18S("Specify the item definitions file"),
I18S("Specify the item definitions file")
},
{
GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */
GWEN_ArgsType_Int, /* type */
"timeout", /* name */
0, /* minnum */
1, /* maxnum */
"T", /* short option */
"timeout", /* long option */
I18S("Specify timeout in second (default: no timeout)"),
I18S("Specify timeout in second (default: no timeout)")
},
{
GWEN_ARGS_FLAGS_HELP | GWEN_ARGS_FLAGS_LAST, /* flags */
GWEN_ArgsType_Int, /* type */
"help", /* name */
0, /* minnum */
0, /* maxnum */
"h", /* short option */
"help",
I18S("Show this help screen."),
I18S("Show this help screen.")
}
};
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;
}