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

Simplified IPC code to use less different IPC messages. Share more code. More qork on MQTT code.

Browse Source
This commit is contained in:
Martin Preuss
2023-10-01 21:31:02 +02:00
parent 0f896c1729
commit 1e27223dfa
50 changed files with 1326 additions and 698 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
aqhome/mqtt/endpoint_mqttc.c
View File

@@ -16,9 +16,12 @@
#include "aqhome/mqtt/msg_mqtt_connect.h"
#include "aqhome/mqtt/msg_mqtt_connack.h"
#include "aqhome/mqtt/msg_mqtt_publish.h"
#include "aqhome/mqtt/msg_mqtt_subscribe.h"
#include "aqhome/mqtt/msg_mqtt_suback.h"
#include <gwenhywfar/endpoint_tcpc.h>
#include <gwenhywfar/endpoint_msgio.h>
#include <gwenhywfar/endpoint_multilayer.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/inherit.h>
@@ -27,6 +30,12 @@
#define AQH_ENDPOINT2_MQTTC_RECONNECT_TIME 5
#define AQH_ENDPOINT2_MQTTC_CONNECT_TIMEOUT 10
#define AQH_ENDPOINT2_MQTTC_STAGE_NONE 0
#define AQH_ENDPOINT2_MQTTC_STAGE_CONNREQ 1
#define AQH_ENDPOINT2_MQTTC_STAGE_SUBREQ 2
#define AQH_ENDPOINT2_MQTTC_STAGE_UP 10
/* ------------------------------------------------------------------------------------------------
@@ -34,22 +43,15 @@
* ------------------------------------------------------------------------------------------------
*/
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);
static int _startConnect(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild);
static void _checkSockets(GWEN_MSG_ENDPOINT *ep,
GWEN_MSG_ENDPOINT *epChild,
GWEN_SOCKETSET *readSet,
GWEN_SOCKETSET *writeSet,
GWEN_SOCKETSET *xSet);
static void _lookForAndHandleConnAck(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild);
static void _lookForAndHandleSubAck(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild);
static int _sendSubscribeMsg(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild, const char *topicFilter);
@@ -67,9 +69,11 @@ GWEN_MSG_ENDPOINT *AQH_MqttClientEndpoint_new(const char *clientId,
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);
ep=GWEN_MultilayerEndpoint_new(name?name:AQH_ENDPOINT2_MQTT_NAME, groupId);
GWEN_MultilayerEndpoint_SetConnectTimeoutInSeconds(ep, AQH_ENDPOINT2_MQTTC_CONNECT_TIMEOUT);
GWEN_MultilayerEndpoint_SetReconnectTimeInSeconds(ep, AQH_ENDPOINT2_MQTTC_RECONNECT_TIME);
GWEN_MultilayerEndpoint_SetStartConnectFn(ep, _startConnect);
GWEN_MultilayerEndpoint_SetCheckSocketsFn(ep, _checkSockets);
epChild=GWEN_TcpcEndpoint_new(host, port, NULL, groupId);
GWEN_MsgIoEndpoint_Extend(epChild);
@@ -82,36 +86,6 @@ GWEN_MSG_ENDPOINT *AQH_MqttClientEndpoint_new(const char *clientId,
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) {
@@ -155,152 +129,52 @@ uint16_t AQH_MqttClientEndpoint_GetNextPacketId(const GWEN_MSG_ENDPOINT *ep)
void _moveMessagesBetweenLists(GWEN_MSG_LIST *srcList, GWEN_MSG_LIST *dstList)
void _checkSockets(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
{
GWEN_MSG *msg;
while( (msg=GWEN_Msg_List_First(srcList)) ) {
GWEN_Msg_List_Del(msg);
GWEN_Msg_List_Add(msg, dstList);
}
}
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 _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));
}
void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
{
GWEN_MSG *msg;
int stage;
GWEN_MsgEndpoint_CheckSockets(epChild, readSet, writeSet, xSet); /* let base layer work */
stage=GWEN_MultilayerEndpoint_GetStage(ep);
switch(stage) {
case AQH_ENDPOINT2_MQTTC_STAGE_CONNREQ: _lookForAndHandleConnAck(ep, epChild); break;
case AQH_ENDPOINT2_MQTTC_STAGE_SUBREQ: _lookForAndHandleSubAck(ep, epChild); break;
default: break;
}
}
int _startConnect(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild)
{
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);
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_CONNREQ);
return rv; /* result from GWEN_TcpcEndpoint_StartConnect() above */
}
}
return GWEN_ERROR_GENERIC;
}
void _lookForAndHandleConnAck(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild)
{
GWEN_MSG *msg;
msg=GWEN_MsgEndpoint_GetFirstReceivedMessage(epChild);
while(msg) {
GWEN_MSG *msgNext;
@@ -308,17 +182,38 @@ void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChi
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);
DBG_INFO(AQH_LOGDOMAIN, "Positive CONNACK response");
if (GWEN_MsgEndpoint_GetFlags(ep) & AQH_ENDPOINT2_MQTTCLIENT_FLAGS_SUBSCRIBEALL) {
int rv;
DBG_INFO(AQH_LOGDOMAIN, "Sending subscribe message");
rv=_sendSubscribeMsg(ep, epChild, "#");
if (rv<0) {
DBG_ERROR(AQH_LOGDOMAIN, "Error sending SUBSCRIBE request (%d)", rv);
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_NONE);
GWEN_MsgEndpoint_Disconnect(epChild);
GWEN_MsgEndpoint_Disconnect(ep);
}
else {
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_SUBREQ);
}
}
else {
DBG_INFO(AQH_LOGDOMAIN, "Connected (no auto-subscription requested)");
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_UP);
GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTED);
}
}
else {
DBG_ERROR(AQH_LOGDOMAIN, "Negative CONNACK response (%d)", code);
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_NONE);
GWEN_MsgEndpoint_Disconnect(epChild);
GWEN_MsgEndpoint_Disconnect(ep);
}
@@ -334,30 +229,64 @@ void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChi
int _startConnect(GWEN_MSG_ENDPOINT *ep)
void _lookForAndHandleSubAck(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild)
{
GWEN_MSG_ENDPOINT *epChild;
GWEN_MSG *msg;
epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
if (epChild) {
int rv;
GWEN_MSG *msg;
msg=GWEN_MsgEndpoint_GetFirstReceivedMessage(epChild);
while(msg) {
GWEN_MSG *msgNext;
uint8_t msgType;
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;
msgNext=GWEN_Msg_List_Next(msg);
msgType=AQH_MqttMsg_GetMsgTypeAndFlags(msg) & 0xf0;
if (msgType==AQH_MQTTMSG_MSGTYPE_SUBACK) {
int code;
GWEN_Msg_List_Del(msg); /* remove from list */
code=AQH_SubAckMqttMsg_GetResultCode(msg);
if (code!=128) {
DBG_INFO(AQH_LOGDOMAIN, "Positive SUBACK response, connected (%d, %02x)", code, code);
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_UP);
GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTED);
}
else {
DBG_ERROR(AQH_LOGDOMAIN, "Negative SUBACK response (%d)", code);
GWEN_MultilayerEndpoint_SetStage(ep, AQH_ENDPOINT2_MQTTC_STAGE_NONE);
GWEN_MsgEndpoint_Disconnect(epChild);
GWEN_MsgEndpoint_Disconnect(ep);
}
GWEN_Msg_free(msg);
break;
}
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 */
else {
DBG_ERROR(AQH_LOGDOMAIN, "Ignoring response (%d)", msgType);
}
msg=msgNext;
} /* while */
}
int _sendSubscribeMsg(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild, const char *topicFilter)
{
uint16_t pckId;
GWEN_MSG *msgOut;
DBG_INFO(AQH_LOGDOMAIN, "Sending SUBSCRIBE %s", topicFilter);
pckId=AQH_MqttClientEndpoint_GetNextPacketId(ep);
msgOut=GWEN_SubscribeMqttMsg_new(AQH_MQTTMSG_MSGTYPE_SUBSCRIBE, pckId, topicFilter, 0);
if (msgOut==NULL) {
DBG_ERROR(AQH_LOGDOMAIN, "Error creating message");
return GWEN_ERROR_INTERNAL;
}
return GWEN_ERROR_GENERIC;
GWEN_MsgEndpoint_AddSendMessage(epChild, msgOut);
return 0;
}