adapted to latest changes in gwen (msgio API v2 becomes v1).
This commit is contained in:
Martin Preuss
392
aqhome/mqtt/endpoint_mqttc.c
Normal file
392
aqhome/mqtt/endpoint_mqttc.c
Normal file
@@ -0,0 +1,392 @@
|
||||
/****************************************************************************
|
||||
* 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 <config.h>
|
||||
#endif
|
||||
|
||||
#include "aqhome/mqtt/endpoint_mqttc.h"
|
||||
|
||||
#include "aqhome/mqtt/endpoint_mqtt.h"
|
||||
#include "aqhome/mqtt/msg_mqtt_connect.h"
|
||||
#include "aqhome/mqtt/msg_mqtt_connack.h"
|
||||
#include "aqhome/mqtt/msg_mqtt_publish.h"
|
||||
|
||||
#include <gwenhywfar/endpoint_tcpc.h>
|
||||
#include <gwenhywfar/endpoint_msgio.h>
|
||||
#include <gwenhywfar/debug.h>
|
||||
#include <gwenhywfar/inherit.h>
|
||||
|
||||
|
||||
#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_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_UNUSED GWEN_SOCKETSET *xSet);
|
||||
static void _checkSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
|
||||
static int _startConnect(GWEN_MSG_ENDPOINT *ep);
|
||||
static void _moveMessagesBetweenLists(GWEN_MSG_LIST *srcList, GWEN_MSG_LIST *dstList);
|
||||
|
||||
static void _addSocketsWhenUnconnected(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
|
||||
static void _addSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
|
||||
static void _addSocketsWhenConnected(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
|
||||
|
||||
static void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
|
||||
static void _checkSocketsWhenConnected(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet);
|
||||
|
||||
|
||||
|
||||
/* ------------------------------------------------------------------------------------------------
|
||||
* implementations
|
||||
* ------------------------------------------------------------------------------------------------
|
||||
*/
|
||||
|
||||
|
||||
|
||||
GWEN_MSG_ENDPOINT *AQH_MqttClientEndpoint_new(const char *clientId,
|
||||
const char *host, int port,
|
||||
const char *name, int groupId)
|
||||
{
|
||||
GWEN_MSG_ENDPOINT *ep;
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
ep=GWEN_MsgEndpoint_new(name?name:AQH_ENDPOINT2_MQTT_NAME, groupId);
|
||||
GWEN_MsgEndpoint_SetAddSocketsFn(ep, _addSockets);
|
||||
GWEN_MsgEndpoint_SetCheckSocketsFn(ep, _checkSockets);
|
||||
|
||||
epChild=GWEN_TcpcEndpoint_new(host, port, NULL, groupId);
|
||||
GWEN_MsgIoEndpoint_Extend(epChild);
|
||||
AQH_MqttEndpoint_Extend(epChild);
|
||||
AQH_MqttEndpoint_SetClientId(epChild, clientId);
|
||||
GWEN_MsgEndpoint_Tree2_AddChild(ep, epChild);
|
||||
|
||||
return ep;
|
||||
}
|
||||
|
||||
|
||||
|
||||
int AQH_MqttClientEndpoint_StartConnect(GWEN_MSG_ENDPOINT *ep)
|
||||
{
|
||||
if (ep) {
|
||||
if (GWEN_MsgEndpoint_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_MsgEndpoint_GetName(ep), rv);
|
||||
GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING);
|
||||
}
|
||||
else {
|
||||
DBG_INFO(AQH_LOGDOMAIN, "Endpoint %s: Connecting.", GWEN_MsgEndpoint_GetName(ep));
|
||||
GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING);
|
||||
}
|
||||
return rv;
|
||||
}
|
||||
else {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Endpoint %s: Not unconnected", GWEN_MsgEndpoint_GetName(ep));
|
||||
}
|
||||
}
|
||||
else {
|
||||
DBG_ERROR(GWEN_LOGDOMAIN, "No endpoint");
|
||||
}
|
||||
return GWEN_ERROR_GENERIC;
|
||||
}
|
||||
|
||||
|
||||
|
||||
uint16_t AQH_MqttClientEndpoint_GetKeepAliveTime(const GWEN_MSG_ENDPOINT *ep)
|
||||
{
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild)
|
||||
return AQH_MqttEndpoint_GetKeepAliveTime(epChild);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
|
||||
void AQH_MqttClientEndpoint_SetKeepAliveTime(GWEN_MSG_ENDPOINT *ep, uint16_t i)
|
||||
{
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild) {
|
||||
AQH_MqttEndpoint_SetKeepAliveTime(epChild, i);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
uint16_t AQH_MqttClientEndpoint_GetNextPacketId(const GWEN_MSG_ENDPOINT *ep)
|
||||
{
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild) {
|
||||
return AQH_MqttEndpoint_GetNextPacketId(epChild);
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
||||
|
||||
const char *AQH_MqttClientEndpoint_GetTopicPrefix(const GWEN_MSG_ENDPOINT *ep)
|
||||
{
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild)
|
||||
return AQH_MqttEndpoint_GetTopicPrefix(epChild);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
|
||||
|
||||
void AQH_MqttClientEndpoint_SetTopicPrefix(GWEN_MSG_ENDPOINT *ep, const char *s)
|
||||
{
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild)
|
||||
AQH_MqttEndpoint_SetTopicPrefix(epChild, s);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
void _addSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild) {
|
||||
if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED)
|
||||
_addSocketsWhenUnconnected(ep, epChild, readSet, writeSet, xSet);
|
||||
else {
|
||||
if (GWEN_MsgEndpoint_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Error on tcp layer, disconnecting");
|
||||
GWEN_MsgEndpoint_Disconnect(epChild);
|
||||
GWEN_MsgEndpoint_Disconnect(ep);
|
||||
}
|
||||
else {
|
||||
if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTING)
|
||||
_addSocketsWhenConnecting(ep, epChild, readSet, writeSet, xSet);
|
||||
if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTED)
|
||||
_addSocketsWhenConnected(ep, epChild, readSet, writeSet, xSet);
|
||||
}
|
||||
}
|
||||
} /* if (epChild) */
|
||||
} /* if (ep) */
|
||||
}
|
||||
|
||||
|
||||
|
||||
void _checkSockets(GWEN_MSG_ENDPOINT *ep, GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
DBG_DEBUG(AQH_LOGDOMAIN, "Checking sockets in state %d", GWEN_MsgEndpoint_GetState(ep));
|
||||
if (ep) {
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild) {
|
||||
if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
|
||||
/* nothing to do here */
|
||||
} /* if GWEN_MSG_ENDPOINT_STATE_UNCONNECTED */
|
||||
else {
|
||||
if (GWEN_MsgEndpoint_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Error on tcp layer, disconnecting");
|
||||
GWEN_MsgEndpoint_Disconnect(epChild);
|
||||
GWEN_MsgEndpoint_Disconnect(ep);
|
||||
}
|
||||
else {
|
||||
if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTING)
|
||||
_checkSocketsWhenConnecting(ep, epChild, readSet, writeSet, xSet);
|
||||
else if (GWEN_MsgEndpoint_GetState(ep)==GWEN_MSG_ENDPOINT_STATE_CONNECTED)
|
||||
_checkSocketsWhenConnected(ep, epChild, readSet, writeSet, xSet);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
void _addSocketsWhenUnconnected(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
time_t now;
|
||||
|
||||
now=time(NULL);
|
||||
if ((now-GWEN_MsgEndpoint_GetTimeOfLastStateChange(ep))>=AQH_ENDPOINT2_MQTTC_RECONNECT_TIME) {
|
||||
int rv;
|
||||
|
||||
/* (re)connect, set state */
|
||||
DBG_INFO(AQH_LOGDOMAIN, "Starting to (re-)connect");
|
||||
rv=AQH_MqttClientEndpoint_StartConnect(ep);
|
||||
if (rv<0 && rv!=GWEN_ERROR_IN_PROGRESS) {
|
||||
DBG_INFO(GWEN_LOGDOMAIN, "here (%d)", rv);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
void _addSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
time_t now;
|
||||
|
||||
now=time(NULL);
|
||||
if ((now-GWEN_MsgEndpoint_GetTimeOfLastStateChange(ep))>=AQH_ENDPOINT2_MQTTC_CONNECT_TIMEOUT ||
|
||||
GWEN_MsgEndpoint_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Timeout on connect");
|
||||
GWEN_MsgEndpoint_Disconnect(epChild);
|
||||
GWEN_MsgEndpoint_Disconnect(ep);
|
||||
}
|
||||
else
|
||||
GWEN_MsgEndpoint_AddSockets(epChild, readSet, writeSet, xSet);
|
||||
}
|
||||
|
||||
|
||||
|
||||
void _addSocketsWhenConnected(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
if (GWEN_MsgEndpoint_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Error on tcp layer, disconnecting");
|
||||
GWEN_MsgEndpoint_Disconnect(epChild);
|
||||
GWEN_MsgEndpoint_Disconnect(ep);
|
||||
}
|
||||
else {
|
||||
/* move to-send messages to child */
|
||||
_moveMessagesBetweenLists(GWEN_MsgEndpoint_GetSendMessageList(ep), GWEN_MsgEndpoint_GetSendMessageList(epChild));
|
||||
GWEN_MsgEndpoint_AddSockets(epChild, readSet, writeSet, xSet);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
GWEN_MSG *msg;
|
||||
|
||||
GWEN_MsgEndpoint_CheckSockets(epChild, readSet, writeSet, xSet); /* let base layer work */
|
||||
|
||||
msg=GWEN_MsgEndpoint_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_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTED);
|
||||
}
|
||||
else {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Negative CONNACK response (%d)", code);
|
||||
GWEN_MsgEndpoint_Disconnect(epChild);
|
||||
GWEN_MsgEndpoint_Disconnect(ep);
|
||||
}
|
||||
GWEN_Msg_free(msg);
|
||||
break;
|
||||
}
|
||||
else {
|
||||
DBG_ERROR(AQH_LOGDOMAIN, "Ignoring response (%d)", msgType);
|
||||
}
|
||||
msg=msgNext;
|
||||
} /* while */
|
||||
}
|
||||
|
||||
|
||||
|
||||
void _checkSocketsWhenConnected(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
|
||||
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
|
||||
{
|
||||
_moveMessagesBetweenLists(GWEN_MsgEndpoint_GetSendMessageList(ep), GWEN_MsgEndpoint_GetSendMessageList(epChild));
|
||||
GWEN_MsgEndpoint_CheckSockets(epChild, readSet, writeSet, xSet);
|
||||
_moveMessagesBetweenLists(GWEN_MsgEndpoint_GetReceivedMessageList(epChild), GWEN_MsgEndpoint_GetReceivedMessageList(ep));
|
||||
}
|
||||
|
||||
|
||||
|
||||
int _startConnect(GWEN_MSG_ENDPOINT *ep)
|
||||
{
|
||||
GWEN_MSG_ENDPOINT *epChild;
|
||||
|
||||
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
|
||||
if (epChild) {
|
||||
int rv;
|
||||
GWEN_MSG *msg;
|
||||
|
||||
rv=GWEN_TcpcEndpoint_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_MqttEndpoint_CreateMsgConnect(epChild);
|
||||
if (msg) {
|
||||
GWEN_MsgEndpoint_AddSendMessage(epChild, msg);
|
||||
GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING);
|
||||
return rv; /* result from GWEN_TcpcEndpoint_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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user