562 lines
11 KiB
C
562 lines
11 KiB
C
/****************************************************************************
|
|
* 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 <config.h>
|
|
#endif
|
|
|
|
#include "aqhome/msgendpoint_p.h"
|
|
|
|
|
|
#include <gwenhywfar/misc.h>
|
|
#include <gwenhywfar/list.h>
|
|
#include <gwenhywfar/error.h>
|
|
#include <gwenhywfar/debug.h>
|
|
#include <gwenhywfar/text.h>
|
|
|
|
#include <stdio.h>
|
|
#include <fcntl.h>
|
|
#include <termios.h>
|
|
#include <unistd.h>
|
|
#include <errno.h>
|
|
#include <sys/ioctl.h>
|
|
|
|
#include <stdlib.h>
|
|
#include <sys/time.h>
|
|
#include <sys/types.h>
|
|
#include <unistd.h>
|
|
|
|
#include <sys/types.h>
|
|
#include <sys/socket.h>
|
|
#include <netinet/in.h>
|
|
#ifdef HAVE_ARPA_INET_H
|
|
# include <arpa/inet.h>
|
|
#endif
|
|
|
|
|
|
#define AQH_MSG_ENDPOINT_BUFFERSIZE 32
|
|
|
|
|
|
GWEN_LIST_FUNCTIONS(AQH_MSG_ENDPOINT, AQH_MsgEndpoint)
|
|
GWEN_INHERIT_FUNCTIONS(AQH_MSG_ENDPOINT)
|
|
|
|
|
|
static int _internalHandleReadable(AQH_MSG_ENDPOINT *ep);
|
|
static int _internalHandleWritable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr);
|
|
static int _setSocketNonBlocking(int fd);
|
|
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT *AQH_MsgEndpoint_new(int fd, int groupId)
|
|
{
|
|
AQH_MSG_ENDPOINT *ep;
|
|
|
|
GWEN_NEW_OBJECT(AQH_MSG_ENDPOINT, ep);
|
|
GWEN_INHERIT_INIT(AQH_MSG_ENDPOINT, ep);
|
|
GWEN_LIST_INIT(AQH_MSG_ENDPOINT, ep);
|
|
ep->fd=fd;
|
|
ep->groupId=groupId;
|
|
ep->receivedMessageList=AQH_Msg_List_new();
|
|
ep->sendMessageList=AQH_Msg_List_new();
|
|
|
|
return ep;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_free(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
if (ep) {
|
|
GWEN_LIST_FINI(AQH_MSG_ENDPOINT, ep);
|
|
GWEN_INHERIT_FINI(AQH_MSG_ENDPOINT, ep);
|
|
if (ep->fd>=0)
|
|
close(ep->fd);
|
|
AQH_Msg_free(ep->currentlyReceivedMsg);
|
|
AQH_Msg_List_free(ep->receivedMessageList);
|
|
AQH_Msg_List_free(ep->sendMessageList);
|
|
GWEN_FREE_OBJECT(ep);
|
|
}
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_GetFd(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->fd;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_SetFd(AQH_MSG_ENDPOINT *ep, int fd)
|
|
{
|
|
ep->fd=fd;
|
|
}
|
|
|
|
|
|
uint32_t AQH_MsgEndpoint_GetGroupId(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->groupId;
|
|
}
|
|
|
|
|
|
|
|
uint32_t AQH_MsgEndpoint_GetAcceptedMsgGroups(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->acceptedMsgGroups;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_SetAcceptedMsgGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->acceptedMsgGroups=f;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_AddAcceptedMsgGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->acceptedMsgGroups|=f;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_DelAcceptedMsgGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->acceptedMsgGroups&=~f;
|
|
}
|
|
|
|
|
|
|
|
uint32_t AQH_MsgEndpoint_GetAcceptedEndpointGroups(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->acceptedEndpointGroups;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_SetAcceptedEndpointGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->acceptedEndpointGroups=f;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_AddAcceptedEndpointGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->acceptedEndpointGroups|=f;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_DelAcceptedEndpointGroups(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->acceptedEndpointGroups&=~f;
|
|
}
|
|
|
|
|
|
|
|
uint32_t AQH_MsgEndpoint_GetFlags(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->flags;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_SetFlags(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->flags=f;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_AddFlags(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->flags|=f;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_SubFlags(AQH_MSG_ENDPOINT *ep, uint32_t f)
|
|
{
|
|
ep->flags&=~f;
|
|
}
|
|
|
|
|
|
|
|
|
|
AQH_MSG_LIST *AQH_MsgEndpoint_GetReceivedMessageList(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->receivedMessageList;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG *AQH_MsgEndpoint_TakeFirstReceivedMessage(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
AQH_MSG *msg;
|
|
|
|
msg=AQH_Msg_List_First(ep->receivedMessageList);
|
|
if (msg)
|
|
AQH_Msg_List_Del(msg);
|
|
return msg;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_LIST *AQH_MsgEndpoint_GetSendMessageList(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->sendMessageList;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_AddReceivedMessage(AQH_MSG_ENDPOINT *ep, AQH_MSG *m)
|
|
{
|
|
AQH_Msg_List_Add(m, ep->receivedMessageList);
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_AddSendMessage(AQH_MSG_ENDPOINT *ep, AQH_MSG *m)
|
|
{
|
|
AQH_Msg_List_Add(m, ep->sendMessageList);
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG *AQH_MsgEndpoint_GetCurrentlyReceivedMsg(const AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return ep->currentlyReceivedMsg;
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_SetCurrentlyReceivedMsg(AQH_MSG_ENDPOINT *ep, AQH_MSG *m)
|
|
{
|
|
AQH_Msg_free(ep->currentlyReceivedMsg);
|
|
ep->currentlyReceivedMsg=m;
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_HandleReadable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr)
|
|
{
|
|
return (ep->handleReadableFn)?(ep->handleReadableFn(ep, emgr)):_internalHandleReadable(ep);
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_HandleWritable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr)
|
|
{
|
|
return (ep->handleWritableFn)?(ep->handleWritableFn(ep, emgr)):_internalHandleWritable(ep, emgr);
|
|
}
|
|
|
|
|
|
|
|
void AQH_MsgEndpoint_Run(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
if (ep->runFn)
|
|
ep->runFn(ep);
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_StartMsg(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
int rv;
|
|
|
|
rv=(ep->startMsgFn)?(ep->startMsgFn(ep)):0;
|
|
if (rv==0)
|
|
ep->sendingMessage=1;
|
|
return rv;
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_EndMsg(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
int rv;
|
|
|
|
rv=(ep->endMsgFn)?(ep->endMsgFn(ep)):0;
|
|
if (rv==0)
|
|
ep->sendingMessage=0;
|
|
return rv;
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_CheckMsg(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return (ep->checkMsgFn)?(ep->checkMsgFn(ep)):1;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_HANDLEREADABLE_FN AQH_MsgEndpoint_SetHandleReadableFn(AQH_MSG_ENDPOINT *ep,
|
|
AQH_MSG_ENDPOINT_HANDLEREADABLE_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_HANDLEREADABLE_FN oldFn;
|
|
|
|
oldFn=ep->handleReadableFn;
|
|
ep->handleReadableFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_HANDLEWRITABLE_FN AQH_MsgEndpoint_SetHandleWritableFn(AQH_MSG_ENDPOINT *ep,
|
|
AQH_MSG_ENDPOINT_HANDLEWRITABLE_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_HANDLEWRITABLE_FN oldFn;
|
|
|
|
oldFn=ep->handleWritableFn;
|
|
ep->handleWritableFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_GET_READFD_FN AQH_MsgEndpoint_SetGetReadFdFn(AQH_MSG_ENDPOINT *ep,
|
|
AQH_MSG_ENDPOINT_GET_READFD_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_GET_READFD_FN oldFn;
|
|
|
|
oldFn=ep->getReadFdFn;
|
|
ep->getReadFdFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_GET_WRITEFD_FN AQH_MsgEndpoint_SetGetWriteFdFn(AQH_MSG_ENDPOINT *ep,
|
|
AQH_MSG_ENDPOINT_GET_WRITEFD_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_GET_WRITEFD_FN oldFn;
|
|
|
|
oldFn=ep->getWriteFdFn;
|
|
ep->getWriteFdFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_RUN_FN AQH_MsgEndpoint_SetRunFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_RUN_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_RUN_FN oldFn;
|
|
|
|
oldFn=ep->runFn;
|
|
ep->runFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_STARTMSG_FN AQH_MsgEndpoint_SetStartMsgFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_STARTMSG_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_STARTMSG_FN oldFn;
|
|
|
|
oldFn=ep->startMsgFn;
|
|
ep->startMsgFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_ENDMSG_FN AQH_MsgEndpoint_SetEndMsgFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_ENDMSG_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_ENDMSG_FN oldFn;
|
|
|
|
oldFn=ep->endMsgFn;
|
|
ep->endMsgFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
AQH_MSG_ENDPOINT_CHECKMSG_FN AQH_MsgEndpoint_SetCheckMsgFn(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_CHECKMSG_FN f)
|
|
{
|
|
AQH_MSG_ENDPOINT_CHECKMSG_FN oldFn;
|
|
|
|
oldFn=ep->checkMsgFn;
|
|
ep->checkMsgFn=f;
|
|
return oldFn;
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_GetReadFd(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
return (ep->getReadFdFn)?(ep->getReadFdFn(ep)):(ep->fd);
|
|
}
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_GetWriteFd(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
if (ep->getWriteFdFn)
|
|
return ep->getWriteFdFn(ep);
|
|
else {
|
|
int somethingToWrite;
|
|
|
|
somethingToWrite=(AQH_Msg_List_First(ep->sendMessageList)!=NULL)?1:0;
|
|
if (somethingToWrite)
|
|
return ep->fd;
|
|
}
|
|
return GWEN_ERROR_NO_DATA;
|
|
}
|
|
|
|
|
|
|
|
|
|
int AQH_MsgEndpoint_DiscardInput(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
int rv;
|
|
uint8_t buffer[AQH_MSG_ENDPOINT_BUFFERSIZE];
|
|
|
|
do {
|
|
rv=read(ep->fd, buffer, sizeof(buffer));
|
|
} while( (rv>0 || (rv<0) && errno==EINTR));
|
|
if (rv<0 && errno!=EAGAIN && errno!=EWOULDBLOCK) {
|
|
DBG_ERROR(NULL, "Error on read(): %s (%d)", strerror(errno), errno);
|
|
return GWEN_ERROR_IO;
|
|
}
|
|
else if (rv==0) {
|
|
DBG_ERROR(NULL, "EOF met on read()");
|
|
return GWEN_ERROR_IO;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
|
|
|
|
int _internalHandleReadable(AQH_MSG_ENDPOINT *ep)
|
|
{
|
|
int rv;
|
|
uint8_t buffer[AQH_MSG_ENDPOINT_BUFFERSIZE*2];
|
|
int len;
|
|
int i;
|
|
|
|
do {
|
|
rv=read(ep->fd, buffer, sizeof(buffer));
|
|
} while( (rv<0) && errno==EINTR);
|
|
if (rv<0) {
|
|
if (errno==EAGAIN || errno==EWOULDBLOCK)
|
|
return GWEN_ERROR_TRY_AGAIN;
|
|
DBG_ERROR(NULL, "Error on read(): %s (%d)", strerror(errno), errno);
|
|
return GWEN_ERROR_IO;
|
|
}
|
|
else if (rv==0) {
|
|
DBG_ERROR(NULL, "EOF met on read()");
|
|
return GWEN_ERROR_IO;
|
|
}
|
|
len=rv;
|
|
|
|
for (i=0; i<len; i++) {
|
|
if (ep->currentlyReceivedMsg==NULL)
|
|
ep->currentlyReceivedMsg=AQH_Msg_new();
|
|
rv=AQH_Msg_AddByte(ep->currentlyReceivedMsg, buffer[i]);
|
|
if (rv<0) {
|
|
DBG_ERROR(NULL, "here (%d)", rv);
|
|
return rv;
|
|
}
|
|
rv=AQH_Msg_IsMsgComplete(ep->currentlyReceivedMsg);
|
|
if (rv<0) {
|
|
/* invalid message */
|
|
DBG_ERROR(NULL, "Invalid message, discarding");
|
|
AQH_Msg_free(ep->currentlyReceivedMsg);
|
|
ep->currentlyReceivedMsg=NULL;
|
|
rv=AQH_MsgEndpoint_DiscardInput(ep);
|
|
if (rv<0) {
|
|
DBG_ERROR(NULL, "here (%d)", rv);
|
|
return rv;
|
|
}
|
|
}
|
|
else if (rv>0) {
|
|
if (!AQH_Msg_IsChecksumValid(ep->currentlyReceivedMsg)) {
|
|
DBG_ERROR(NULL, "Invalid checksum, discarding message");
|
|
GWEN_Text_DumpString(AQH_Msg_GetBuffer(ep->currentlyReceivedMsg), AQH_Msg_GetBytesInBuffer(ep->currentlyReceivedMsg), 6);
|
|
AQH_Msg_free(ep->currentlyReceivedMsg);
|
|
ep->currentlyReceivedMsg=NULL;
|
|
rv=AQH_MsgEndpoint_DiscardInput(ep);
|
|
if (rv<0) {
|
|
DBG_ERROR(NULL, "here (%d)", rv);
|
|
return rv;
|
|
}
|
|
}
|
|
else {
|
|
/* valid msg received, add to list */
|
|
AQH_Msg_List_Add(ep->currentlyReceivedMsg, ep->receivedMessageList);
|
|
ep->currentlyReceivedMsg=NULL;
|
|
}
|
|
}
|
|
}
|
|
|
|
return 0;
|
|
}
|
|
|
|
|
|
|
|
int _internalHandleWritable(AQH_MSG_ENDPOINT *ep, AQH_MSG_ENDPOINT_MGR *emgr)
|
|
{
|
|
AQH_MSG *msg;
|
|
|
|
msg=AQH_Msg_List_First(ep->sendMessageList);
|
|
if (msg) {
|
|
uint8_t pos;
|
|
int len;
|
|
int remaining;
|
|
int rv;
|
|
|
|
rv=AQH_MsgEndpoint_CheckMsg(ep);
|
|
if (rv<0 || rv==1) {
|
|
DBG_ERROR(NULL, "Line busy, not sending");
|
|
usleep(100);
|
|
return 0;
|
|
}
|
|
pos=AQH_Msg_GetCurrentPos(msg);
|
|
remaining=AQH_Msg_GetRemainingBytes(msg);
|
|
if (remaining>0) {
|
|
const uint8_t *buf;
|
|
|
|
rv=AQH_MsgEndpoint_StartMsg(ep);
|
|
if (rv<0) {
|
|
DBG_ERROR(NULL, "here (%d)", rv);
|
|
return rv;
|
|
}
|
|
|
|
buf=AQH_Msg_GetBuffer(msg)+pos;
|
|
do {
|
|
rv=write(ep->fd, buf, remaining);
|
|
} while(rv<0 && errno==EINTR);
|
|
if (rv<0) {
|
|
if (errno==EAGAIN || errno==EWOULDBLOCK)
|
|
return GWEN_ERROR_TRY_AGAIN;
|
|
DBG_ERROR(NULL, "Error on write(): %s (%d)", strerror(errno), errno);
|
|
return GWEN_ERROR_IO;
|
|
}
|
|
AQH_Msg_IncCurrentPos(msg, rv);
|
|
if (rv==remaining) {
|
|
rv=AQH_MsgEndpoint_EndMsg(ep);
|
|
// TODO: callback msg sent
|
|
AQH_Msg_List_Del(msg);
|
|
AQH_Msg_free(msg);
|
|
if (rv<0) {
|
|
DBG_ERROR(NULL, "here (%d)", rv);
|
|
return rv;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
|
|
|