endpoint_multilayer.c

Go to the documentation of this file.

/****************************************************************************
 * 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

/*#define DISABLE_DEBUGLOG*/


#include "./endpoint_multilayer_p.h"

#include <gwenhywfar/endpoint.h>
#include <gwenhywfar/debug.h>
#include <gwenhywfar/inherit.h>


#define GWEN_ENDPOINT_MULTILAYER_NAME             "multilayer"
#define GWEN_ENDPOINT_MULTILAYER_DEFAULT_RECONNECT_TIME  5
#define GWEN_ENDPOINT_MULTILAYER_DEFAULT_CONNECT_TIMEOUT 10


GWEN_INHERIT(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER);


/* ------------------------------------------------------------------------------------------------
 * forward declarations
 * ------------------------------------------------------------------------------------------------
 */

static void GWENHYWFAR_CB _freeData(void *bp, void *p);

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 _checkForNeedToReconnect(GWEN_MSG_ENDPOINT *ep);
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 *GWEN_MultilayerEndpoint_new(const char *name, int groupId)
{
  GWEN_MSG_ENDPOINT *ep;
  GWEN_ENDPOINT_MULTILAYER *xep;

  ep=GWEN_MsgEndpoint_new(name?name:GWEN_ENDPOINT_MULTILAYER_NAME, groupId);
  GWEN_NEW_OBJECT(GWEN_ENDPOINT_MULTILAYER, xep);
  GWEN_INHERIT_SETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep, xep, _freeData);

  GWEN_MsgEndpoint_SetAddSocketsFn(ep, _addSockets);
  GWEN_MsgEndpoint_SetCheckSocketsFn(ep, _checkSockets);

  xep->reconnectTimeInSeconds=GWEN_ENDPOINT_MULTILAYER_DEFAULT_RECONNECT_TIME;
  xep->connectTimeoutInSeconds=GWEN_ENDPOINT_MULTILAYER_DEFAULT_CONNECT_TIMEOUT;

  return ep;
}



void GWENHYWFAR_CB _freeData(GWEN_UNUSED void *bp, void *p)
{
  GWEN_ENDPOINT_MULTILAYER *xep;

  xep=(GWEN_ENDPOINT_MULTILAYER*) p;
  GWEN_FREE_OBJECT(xep);
}



int GWEN_MultilayerEndpoint_GetReconnectTimeInSeconds(const GWEN_MSG_ENDPOINT *ep)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep)
      return xep->reconnectTimeInSeconds;
  }
  return GWEN_ENDPOINT_MULTILAYER_DEFAULT_RECONNECT_TIME;
}



void GWEN_MultilayerEndpoint_SetReconnectTimeInSeconds(GWEN_MSG_ENDPOINT *ep, int i)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep)
      xep->reconnectTimeInSeconds=i;
  }
}



int GWEN_MultilayerEndpoint_GetConnectTimeoutInSeconds(const GWEN_MSG_ENDPOINT *ep)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep)
      return xep->connectTimeoutInSeconds;
  }
  return GWEN_ENDPOINT_MULTILAYER_DEFAULT_CONNECT_TIMEOUT;
}



void GWEN_MultilayerEndpoint_SetConnectTimeoutInSeconds(GWEN_MSG_ENDPOINT *ep, int i)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep)
      xep->connectTimeoutInSeconds=i;
  }
}



int GWEN_MultilayerEndpoint_GetStage(const GWEN_MSG_ENDPOINT *ep)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep)
      return xep->stage;
  }
  return 0;
}



void GWEN_MultilayerEndpoint_SetStage(GWEN_MSG_ENDPOINT *ep, int i)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep)
      xep->stage=i;
  }
}



GWEN_ENDPOINT_MULTILAYER_STARTCONNECT_FN GWEN_MultilayerEndpoint_SetStartConnectFn(GWEN_MSG_ENDPOINT *ep,
                                                                                   GWEN_ENDPOINT_MULTILAYER_STARTCONNECT_FN fn)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep) {
      GWEN_ENDPOINT_MULTILAYER_STARTCONNECT_FN oldFn;

      oldFn=xep->startConnectFn;
      xep->startConnectFn=fn;
      return oldFn;
    }
  }

  return NULL;
}



GWEN_ENDPOINT_MULTILAYER_CHECKSOCKETS_FN GWEN_MultilayerEndpoint_SetCheckSocketsFn(GWEN_MSG_ENDPOINT *ep,
                                                                                   GWEN_ENDPOINT_MULTILAYER_CHECKSOCKETS_FN fn)
{
  if (ep) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep) {
      GWEN_ENDPOINT_MULTILAYER_CHECKSOCKETS_FN oldFn;

      oldFn=xep->checkSocketsFn;
      xep->checkSocketsFn=fn;
      return oldFn;
    }
  }

  return NULL;
}



int GWEN_MultilayerEndpoint_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(GWEN_LOGDOMAIN, "Endpoint %s: Error connecting (%d)", GWEN_MsgEndpoint_GetName(ep), rv);
        GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING);
      }
      else {
        DBG_INFO(GWEN_LOGDOMAIN, "Endpoint %s: Connecting.", GWEN_MsgEndpoint_GetName(ep));
        GWEN_MsgEndpoint_SetState(ep, GWEN_MSG_ENDPOINT_STATE_CONNECTING);
      }
      return rv;
    }
    else {
      DBG_ERROR(GWEN_LOGDOMAIN, "Endpoint %s: Not unconnected", GWEN_MsgEndpoint_GetName(ep));
    }
  }
  else {
    DBG_ERROR(GWEN_LOGDOMAIN, "No endpoint");
  }
  return GWEN_ERROR_GENERIC;
}



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)
        _checkForNeedToReconnect(ep);
      else {
        if (GWEN_MsgEndpoint_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
          DBG_ERROR(GWEN_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(GWEN_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(GWEN_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 _checkForNeedToReconnect(GWEN_MSG_ENDPOINT *ep)
{
  time_t now;

  now=time(NULL);
  if ((now-GWEN_MsgEndpoint_GetTimeOfLastStateChange(ep))>=GWEN_MultilayerEndpoint_GetReconnectTimeInSeconds(ep)) {
    int rv;

    /* (re)connect, set state */
    DBG_INFO(GWEN_LOGDOMAIN, "Starting to (re-)connect");
    rv=GWEN_MultilayerEndpoint_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))>=GWEN_MultilayerEndpoint_GetConnectTimeoutInSeconds(ep) ||
      GWEN_MsgEndpoint_GetState(epChild)==GWEN_MSG_ENDPOINT_STATE_UNCONNECTED) {
    DBG_ERROR(GWEN_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(GWEN_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));
  _moveMessagesBetweenLists(GWEN_MsgEndpoint_GetReceivedMessageList(epChild), GWEN_MsgEndpoint_GetReceivedMessageList(ep));
  GWEN_MsgEndpoint_CheckSockets(epChild, readSet, writeSet, xSet);
  _moveMessagesBetweenLists(GWEN_MsgEndpoint_GetReceivedMessageList(epChild), GWEN_MsgEndpoint_GetReceivedMessageList(ep));
}



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);
  }
}





int _startConnect(GWEN_MSG_ENDPOINT *ep)
{
  GWEN_MSG_ENDPOINT *epChild;

  epChild=GWEN_MsgEndpoint_Tree2_GetFirstChild(ep);
  if (epChild) {
    GWEN_ENDPOINT_MULTILAYER *xep;

    xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
    if (xep) {
      if (xep->startConnectFn)
        return xep->startConnectFn(ep, epChild);
    }
  }
  return GWEN_ERROR_GENERIC;
}



void _checkSocketsWhenConnecting(GWEN_MSG_ENDPOINT *ep, GWEN_MSG_ENDPOINT *epChild,
                                 GWEN_SOCKETSET *readSet, GWEN_SOCKETSET *writeSet, GWEN_SOCKETSET *xSet)
{
  GWEN_ENDPOINT_MULTILAYER *xep;

  GWEN_MsgEndpoint_CheckSockets(epChild, readSet, writeSet, xSet); /* let base layer work */

  xep=GWEN_INHERIT_GETDATA(GWEN_MSG_ENDPOINT, GWEN_ENDPOINT_MULTILAYER, ep);
  if (xep) {
    if (xep->checkSocketsFn)
      xep->checkSocketsFn(ep, epChild, readSet, writeSet, xSet);
  }
}