/**************************************************************************** * 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 #endif #include "./init.h" #include "./aqhome_mqtt_p.h" #include "./xmlread.h" #include "./xmlwrite.h" #include "aqhome/aqhome.h" #include "aqhome/ipc/endpoint_ipc.h" #include "aqhome/ipc/endpoint_ipcclient.h" #include "aqhome/mqtt/endpoint_mqttc.h" #include #include #include #include #include #include #include #include #include /* ------------------------------------------------------------------------------------------------ * defines * ------------------------------------------------------------------------------------------------ */ #define I18N(msg) msg #define I18S(msg) msg /* ------------------------------------------------------------------------------------------------ * forward declarations * ------------------------------------------------------------------------------------------------ */ static int _createPidFile(const char *pidFilename); static int _setupBroker(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs); static int _setupMqtt(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs); static int _readArgs(int argc, char **argv, GWEN_DB_NODE *dbArgs); /* ------------------------------------------------------------------------------------------------ * implementations * ------------------------------------------------------------------------------------------------ */ int AqHomeMqtt_Init(AQHOME_MQTT *aqh, int argc, char **argv) { int rv; GWEN_DB_NODE *dbArgs; 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"); aqh->dbArgs=dbArgs; aqh->timeout=GWEN_DB_GetIntValue(dbArgs, "timeout", 0, 0); s=GWEN_DB_GetCharValue(dbArgs, "pidfile", 0, AQHOME_MQTT_DEFAULT_PIDFILE); if (s && *s) { AqHomeMqtt_SetPidFile(aqh, s); rv=_createPidFile(s); if (rv<0) { DBG_ERROR(NULL, "Error creating PID file (%d)", rv); return rv; } } s=GWEN_DB_GetCharValue(dbArgs, "devicefile", 0, NULL); if (s && *s) { AqHomeMqtt_SetDeviceFile(aqh, s); } else { GWEN_BUFFER *bufFilename; bufFilename=AQH_GetRuntimeFilePath(AQHOME_MQTT_DEFAULT_DEVICEFILE); if (bufFilename) { AqHomeMqtt_SetDeviceFile(aqh, GWEN_Buffer_GetStart(bufFilename)); GWEN_Buffer_free(bufFilename); } else { DBG_ERROR(NULL, "Could not setup filename for devices, please specify via command line argument"); return GWEN_ERROR_GENERIC; } } aqh->rootEndpoint=GWEN_MsgEndpoint_new("root", 0); rv=_setupMqtt(aqh, dbArgs); if (rv<0) { DBG_ERROR(NULL, "Error setting up connection to broker (%d)", rv); return rv; } rv=_setupBroker(aqh, dbArgs); if (rv<0) { DBG_ERROR(NULL, "Error setting up connection to broker (%d)", rv); return rv; } AqHomeMqtt_LoadRuntimeDeviceFiles(aqh); AqHomeMqtt_ReloadDeviceFiles(aqh); return 0; } void AqHomeMqtt_ReloadDeviceFiles(AQHOME_MQTT *aqh) { AQHMQTT_DEVICE_LIST *deviceList; DBG_ERROR(NULL, "Loading devices description files"); deviceList=AqHomeMqttLog_ReadDataDeviceFiles(aqh); if (deviceList) AqHomeMqtt_SetAvailableDeviceList(aqh, deviceList); } void AqHomeMqtt_LoadRuntimeDeviceFiles(AQHOME_MQTT *aqh) { AQHMQTT_DEVICE_LIST *deviceList; DBG_ERROR(NULL, "Loading registered devices from file \"%s\"", aqh->deviceFile); deviceList=AqHomeMqttLog_ReadDeviceFile(aqh, aqh->deviceFile); if (deviceList) AqHomeMqtt_SetRegisteredDeviceList(aqh, deviceList); } int AqHomeMqtt_SaveRuntimeDeviceFiles(AQHOME_MQTT *aqh) { int rv; rv=AqHomeMqttLog_WriteDevicesFile(aqh, aqh->registeredDeviceList, aqh->deviceFile); if (rv<0) { DBG_INFO(NULL, "Error writing devices to \"%s\" (%d)", aqh->deviceFile, rv); return rv; } return 0; } 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 _setupBroker(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs) { const char *brokerAddress; int brokerPort; const char *brokerClientId; brokerAddress=GWEN_DB_GetCharValue(dbArgs, "brokerAddress", 0, NULL); if (!(brokerAddress && *brokerAddress)) brokerAddress=GWEN_DB_GetCharValue(dbArgs, "ConfigFile/brokerAddress", 0, "127.0.0.1"); brokerPort=GWEN_DB_GetIntValue(dbArgs, "brokerPort", 0, -1); if (brokerPort<0) brokerPort=GWEN_DB_GetIntValue(dbArgs, "ConfigFile/brokerPort", 0, AQHOME_MQTT_DEFAULT_BROKER_PORT); brokerClientId=GWEN_DB_GetCharValue(dbArgs, "brokerClientId", 0, AQHOME_MQTT_DEFAULT_BROKER_CLIENTID); if (brokerAddress && *brokerAddress && brokerPort) { GWEN_MSG_ENDPOINT *ep; GWEN_MSG_ENDPOINT *ipcBaseEndpoint; int rv; ep=AQH_ClientIpcEndpoint_new("brokerIpcClient", 0); ipcBaseEndpoint=AQH_IpcEndpoint_CreateIpcTcpClient(brokerAddress, brokerPort, "brokerPhysEndpoint", 0); AQH_IpcEndpoint_SetServiceName(ipcBaseEndpoint, brokerClientId); GWEN_MsgEndpoint_Tree2_AddChild(ep, ipcBaseEndpoint); GWEN_MsgEndpoint_Tree2_AddChild(aqh->rootEndpoint, ep); aqh->brokerEndpoint=ep; rv=GWEN_MultilayerEndpoint_StartConnect(ep); if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) { DBG_ERROR(NULL, "Error connecting to broker server %s:%d (%d), will retry later", brokerAddress, brokerPort, rv); return rv; } } return 0; } int _setupMqtt(AQHOME_MQTT *aqh, GWEN_DB_NODE *dbArgs) { const char *mqttAddress; int mqttPort; const char *mqttClientId; int mqttKeepAlive; mqttAddress=GWEN_DB_GetCharValue(dbArgs, "mqttAddress", 0, NULL); if (!(mqttAddress && *mqttAddress)) mqttAddress=GWEN_DB_GetCharValue(dbArgs, "ConfigFile/mqttAddr", 0, "127.0.0.1"); mqttPort=GWEN_DB_GetIntValue(dbArgs, "mqttPort", 0, 1883); if (mqttPort<0) mqttPort=GWEN_DB_GetIntValue(dbArgs, "ConfigFile/mqttPort", 0, 1883); mqttClientId=GWEN_DB_GetCharValue(dbArgs, "mqttClientId", 0, "aqhome-mqttlog"); mqttKeepAlive=GWEN_DB_GetIntValue(dbArgs, "mqttKeepAlive", 0, 600); if (mqttAddress && *mqttAddress && mqttPort) { GWEN_MSG_ENDPOINT *epMqtt; DBG_INFO(AQH_LOGDOMAIN, "Connecting to %s (port %d)", mqttAddress, mqttPort); epMqtt=AQH_MqttClientEndpoint_new(mqttClientId, mqttAddress, mqttPort, NULL, 0); if (epMqtt==NULL) { DBG_ERROR(AQH_LOGDOMAIN, "Error creating endpoint TCP"); return GWEN_ERROR_IO; } AQH_MqttClientEndpoint_SetKeepAliveTime(epMqtt, mqttKeepAlive); GWEN_MsgEndpoint_AddFlags(epMqtt, AQH_ENDPOINT2_MQTTCLIENT_FLAGS_SUBSCRIBEALL); GWEN_MsgEndpoint_Tree2_AddChild(aqh->rootEndpoint, epMqtt); aqh->mqttEndpoint=epMqtt; } 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 */ "brokerAddress", /* name */ 0, /* minnum */ 1, /* maxnum */ "ba", /* short option */ "brokeraddress", /* long option */ I18S("Specify the address of the broker server to connect to (disabled if missing)"), I18S("Specify the address of the broker server to connect to (disabled if missing)") }, { GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ GWEN_ArgsType_Int, /* type */ "brokerPort", /* name */ 0, /* minnum */ 1, /* maxnum */ "bp", /* short option */ "brokerport", /* long option */ I18S("Specify the port of the broker server (default: 1899)"), I18S("Specify the port of the broker server (default: 1899)") }, { GWEN_ARGS_FLAGS_HAS_ARGUMENT, /* flags */ GWEN_ArgsType_Char, /* type */ "brokerClientId", /* name */ 0, /* minnum */ 1, /* maxnum */ NULL, /* short option */ "brokerclientid", /* long option */ I18S("Specify client id for the broker server (default: \"nodes\")"), I18S("Specify client id for the broker server (default: \"nodes\")") }, { 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 */ "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_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_HAS_ARGUMENT, /* flags */ GWEN_ArgsType_Char, /* type */ "devicefile", /* name */ 0, /* minnum */ 1, /* maxnum */ "d", /* short option */ "devicefile", /* long option */ I18S("Specify the device file"), I18S("Specify the device file") }, { 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; }