/**************************************************************************** * This file is part of the project Gwenhywfar. * Gwenhywfar (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 "aqhome/mqtt/endpoint2_mqttc.h" #include "aqhome/mqtt/endpoint2_mqtt.h" #include "aqhome/mqtt/msg_mqtt_connect.h" #include "aqhome/mqtt/msg_mqtt_connack.h" #include "aqhome/mqtt/msg_mqtt_publish.h" #include #include #include #include #define AQH_ENDPOINT2_MQTT_NAME "mqtt-client" #define AQH_ENDPOINT2_MQTTC_RECONNECT_TIME 5 #define AQH_ENDPOINT2_MQTTC_CONNECT_TIMEOUT 10 /* ------------------------------------------------------------------------------------------------ * forward declarations * ------------------------------------------------------------------------------------------------ */ static void _addSockets(GWEN_MSG_ENDPOINT2 *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_UNUSED GWEN_SOCKETSET *xSet); static void _checkSockets(GWEN_MSG_ENDPOINT2 *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet); static int _startConnect(GWEN_MSG_ENDPOINT2 *ep); static void _moveMessagesBetweenLists(GWEN_MSG_LIST *srcList, GWEN_MSG_LIST *dstList); static void _addSocketsWhenUnconnected(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet); static void _addSocketsWhenConnecting(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet); static void _addSocketsWhenConnected(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet); static void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet); static void _checkSocketsWhenConnected(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet); /* ------------------------------------------------------------------------------------------------ * implementations * ------------------------------------------------------------------------------------------------ */ GWEN_MSG_ENDPOINT2 *AQH_MqttClientEndpoint2_new(const char *clientId, const char *host, int port, const char *name, int groupId) { GWEN_MSG_ENDPOINT2 *ep; GWEN_MSG_ENDPOINT2 *epChild; ep=GWEN_MsgEndpoint2_new(name?name:AQH_ENDPOINT2_MQTT_NAME, groupId); GWEN_MsgEndpoint2_SetAddSocketsFn(ep, _addSockets); GWEN_MsgEndpoint2_SetCheckSocketsFn(ep, _checkSockets); epChild=GWEN_TcpcEndpoint2_new(host, port, NULL, groupId); GWEN_MsgIoEndpoint2_Extend(epChild); AQH_MqttEndpoint2_Extend(epChild); AQH_MqttEndpoint2_SetClientId(epChild, clientId); GWEN_MsgEndpoint2_Tree2_AddChild(ep, epChild); return ep; } int AQH_MqttClientEndpoint2_StartConnect(GWEN_MSG_ENDPOINT2 *ep) { if (ep) { if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) { int rv; /* connect, set state */ rv=_startConnect(ep); if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) { DBG_INFO(AQH_LOGDOMAIN, "Endpoint %s: Error connecting (%d)", GWEN_MsgEndpoint2_GetName(ep), rv); GWEN_MsgEndpoint2_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING); } else { DBG_INFO(AQH_LOGDOMAIN, "Endpoint %s: Connecting.", GWEN_MsgEndpoint2_GetName(ep)); GWEN_MsgEndpoint2_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING); } return rv; } else { DBG_ERROR(AQH_LOGDOMAIN, "Endpoint %s: Not unconnected", GWEN_MsgEndpoint2_GetName(ep)); } } else { DBG_ERROR(GWEN_LOGDOMAIN, "No endpoint"); } return GWEN_ERROR_GENERIC; } uint16_t AQH_MqttClientEndpoint2_GetKeepAliveTime(const GWEN_MSG_ENDPOINT2 *ep) { if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) return AQH_MqttEndpoint2_GetKeepAliveTime(epChild); } return 0; } void AQH_MqttClientEndpoint2_SetKeepAliveTime(GWEN_MSG_ENDPOINT2 *ep, uint16_t i) { if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) { AQH_MqttEndpoint2_SetKeepAliveTime(epChild, i); } } } uint16_t AQH_MqttClientEndpoint2_GetNextPacketId(const GWEN_MSG_ENDPOINT2 *ep) { if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) { return AQH_MqttEndpoint2_GetNextPacketId(epChild); } } return 0; } const char *AQH_MqttClientEndpoint2_GetTopicPrefix(const GWEN_MSG_ENDPOINT2 *ep) { if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) return AQH_MqttEndpoint2_GetTopicPrefix(epChild); } return NULL; } void AQH_MqttClientEndpoint2_SetTopicPrefix(GWEN_MSG_ENDPOINT2 *ep, const char *s) { if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) AQH_MqttEndpoint2_SetTopicPrefix(epChild, s); } } void _addSockets(GWEN_MSG_ENDPOINT2 *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) { if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) _addSocketsWhenUnconnected(ep, epChild, readSet, writeSet, xSet); else { if (GWEN_MsgEndpoint2_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) { DBG_ERROR(AQH_LOGDOMAIN, "Error on tcp layer, disconnecting"); GWEN_MsgEndpoint2_Disconnect(epChild); GWEN_MsgEndpoint2_Disconnect(ep); } else { if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTING) _addSocketsWhenConnecting(ep, epChild, readSet, writeSet, xSet); if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTED) _addSocketsWhenConnected(ep, epChild, readSet, writeSet, xSet); } } } /* if (epChild) */ } /* if (ep) */ } void _checkSockets(GWEN_MSG_ENDPOINT2 *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { DBG_ERROR(AQH_LOGDOMAIN, "Checking sockets in state %d", GWEN_MsgEndpoint2_GetState(ep)); if (ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) { if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) { /* nothing to do here */ } /* if GWEN_MSG_ENDPOINT_STATE_UNCONNECTED */ else { if (GWEN_MsgEndpoint2_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) { DBG_ERROR(AQH_LOGDOMAIN, "Error on tcp layer, disconnecting"); GWEN_MsgEndpoint2_Disconnect(epChild); GWEN_MsgEndpoint2_Disconnect(ep); } else { if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTING) _checkSocketsWhenConnecting(ep, epChild, readSet, writeSet, xSet); else if (GWEN_MsgEndpoint2_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTED) _checkSocketsWhenConnected(ep, epChild, readSet, writeSet, xSet); } } } } } void _addSocketsWhenUnconnected(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { time_t now; now=time(NULL); if ((now-GWEN_MsgEndpoint2_GetTimeOfLastStateChange(ep))>=AQH_ENDPOINT2_MQTTC_RECONNECT_TIME) { int rv; /* (re)connect, set state */ DBG_INFO(AQH_LOGDOMAIN, "Starting to (re-)connect"); rv=AQH_MqttClientEndpoint2_StartConnect(ep); if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) { DBG_INFO(GWEN_LOGDOMAIN, "here (%d)", rv); } } } void _addSocketsWhenConnecting(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { time_t now; now=time(NULL); if ((now-GWEN_MsgEndpoint2_GetTimeOfLastStateChange(ep))>=AQH_ENDPOINT2_MQTTC_CONNECT_TIMEOUT || GWEN_MsgEndpoint2_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) { DBG_ERROR(AQH_LOGDOMAIN, "Timeout on connect"); GWEN_MsgEndpoint2_Disconnect(epChild); GWEN_MsgEndpoint2_Disconnect(ep); } else GWEN_MsgEndpoint2_AddSockets(epChild, readSet, writeSet, xSet); } void _addSocketsWhenConnected(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { if (GWEN_MsgEndpoint2_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) { DBG_ERROR(AQH_LOGDOMAIN, "Error on tcp layer, disconnecting"); GWEN_MsgEndpoint2_Disconnect(epChild); GWEN_MsgEndpoint2_Disconnect(ep); } else { /* move to-send messages to child */ _moveMessagesBetweenLists(GWEN_MsgEndpoint2_GetSendMessageList(ep), GWEN_MsgEndpoint2_GetSendMessageList(epChild)); GWEN_MsgEndpoint2_AddSockets(epChild, readSet, writeSet, xSet); } } void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { GWEN_MSG *msg; GWEN_MsgEndpoint2_CheckSockets(epChild, readSet, writeSet, xSet); /* let base layer work */ msg=GWEN_MsgEndpoint2_GetFirstReceivedMessage(epChild); while(msg) { GWEN_MSG *msgNext; uint8_t msgType; msgNext=GWEN_Msg_List_Next(msg); msgType=AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0; if (msgType==AQH_MQTTMSG_MSGTYPE_CONNACK) { int code; GWEN_Msg_List_Del(msg); /* remove from list */ code=AQH_ConnAckMqttMsg_GetResultCode(msg); if (code==AQH_MQTTMSG_CONNACK_RESULT_ACCEPTED) { DBG_INFO(AQH_LOGDOMAIN, "Positive CONNACK response, connected"); GWEN_MsgEndpoint2_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTED); } else { DBG_ERROR(AQH_LOGDOMAIN, "Negative CONNACK response (%d)", code); GWEN_MsgEndpoint2_Disconnect(epChild); GWEN_MsgEndpoint2_Disconnect(ep); } GWEN_Msg_free(msg); break; } else { DBG_ERROR(AQH_LOGDOMAIN, "Ignoring response (%d)", msgType); } msg=msgNext; } /* while */ } void _checkSocketsWhenConnected(GWEN_MSG_ENDPOINT2 *ep, GWEN_MSG_ENDPOINT2 *epChild, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet) { _moveMessagesBetweenLists(GWEN_MsgEndpoint2_GetSendMessageList(ep), GWEN_MsgEndpoint2_GetSendMessageList(epChild)); GWEN_MsgEndpoint2_CheckSockets(epChild, readSet, writeSet, xSet); _moveMessagesBetweenLists(GWEN_MsgEndpoint2_GetReceivedMessageList(epChild), GWEN_MsgEndpoint2_GetReceivedMessageList(ep)); } int _startConnect(GWEN_MSG_ENDPOINT2 *ep) { GWEN_MSG_ENDPOINT2 *epChild; epChild=GWEN_MsgEndpoint2_Tree2_GetFirstChild(ep); if (epChild) { int rv; GWEN_MSG *msg; rv=GWEN_TcpcEndpoint2_StartConnect(epChild); if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) { DBG_INFO(AQH_LOGDOMAIN, "Error starting to connect child layer (%d)", rv); return rv; } msg=AQH_MqttEndpoint2_CreateMsgConnect(epChild); if (msg) { GWEN_MsgEndpoint2_AddSendMessage(epChild, msg); GWEN_MsgEndpoint2_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING); return rv; /* result from GWEN_TcpcEndpoint2_StartConnect() above */ } } return GWEN_ERROR_GENERIC; } void _moveMessagesBetweenLists(GWEN_MSG_LIST *srcList, GWEN_MSG_LIST *dstList) { GWEN_MSG *msg; while( (msg=GWEN_Msg_List_First(srcList)) ) { GWEN_Msg_List_Del(msg); GWEN_Msg_List_Add(msg, dstList); } }