testbench: add test for receiver abort and restart

Also
- much improved testbench framework and tools to support
  this type of test. Done some general improvements to the
  tooling.
- imported rsyslog tool "chkseq" for checking
- added more error messages to code
  still not fully done and probably needs some review. but
  even for testbench runs the current set of error messages
  was too small to be useful.
  see also rsyslog#130

configure.ac

@@ -3,8 +3,11 @@
 
 AC_PREREQ(2.61)
 AC_INIT([librelp], [1.2.19.master], [[email protected]])
-AM_INIT_AUTOMAKE
-AM_INIT_AUTOMAKE
+
+# change to the one below if Travis has a timeout
+#AM_INIT_AUTOMAKE([subdir-objects serial-tests])
+AM_INIT_AUTOMAKE([subdir-objects])
+
 m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])])
 AC_CONFIG_SRCDIR([src/relp.c])
 AC_CONFIG_HEADER([config.h])

src/copen.c

@@ -143,7 +143,6 @@ BEGINcommand(S, Init)
 		relpSessSendResponse(pSess, pFrame->txnr, replymsg, sizeof(replymsg) - 1);
 		ABORT_FINALIZE(RELP_RET_SESSION_OPEN);
 	}
-
 	CHKRet(relpOffersConstructFromFrame(&pCltOffers, pFrame));
 	CHKRet(selectOffers(pSess, pCltOffers, &pSrvOffers));
 

src/relpsess.c

@@ -51,6 +51,7 @@
 #include "relp.h"
 #include "relpsess.h"
 #include "relpframe.h"
+#include "relpclt.h"
 #include "sendq.h"
 #include "offers.h"
 #include "dbllinklist.h"
@@ -67,25 +68,24 @@ callOnErr(const relpSess_t *__restrict__ const pThis,
 	const relpRetVal ecode)
 {
 	char objinfo[1024];
+	relpTcp_t *const pTcp = pThis->pTcp;
 	//pThis->pEngine->dbgprint("librelp: generic error: ecode %d, "
 		//"emsg '%s'\n", ecode, emsg);
 	if(pThis->pEngine->onErr != NULL) {
-#if 0 // TODO: FIXME
-		if(pThis->pSrv == NULL) { /* client */
+		if(pTcp->pSrv == NULL) { /* client */
 			snprintf(objinfo, sizeof(objinfo), "conn to srvr %s:%s",
-				 pThis->pClt->pSess->srvAddr,
-				 pThis->pClt->pSess->srvPort);
-		} else if(pThis->pRemHostIP == NULL) { /* server listener */
+				 pTcp->pClt->pSess->srvAddr,
+				 pTcp->pClt->pSess->srvPort);
+		} else if(pTcp->pRemHostIP == NULL) { /* server listener */
 			snprintf(objinfo, sizeof(objinfo), "lstn %s",
-				 pThis->pSrv->pLstnPort);
+				 pTcp->pSrv->pLstnPort);
 		} else { /* server connection to client */
 			snprintf(objinfo, sizeof(objinfo), "lstn %s: conn to clt %s/%s",
-				 pThis->pSrv->pLstnPort, pThis->pRemHostIP,
-				 pThis->pRemHostName);
+				 pTcp->pSrv->pLstnPort, pTcp->pRemHostIP,
+				 pTcp->pRemHostName);
 		}
 		objinfo[sizeof(objinfo)-1] = '\0';
-#endif
-		pThis->pEngine->onErr(pThis->pUsr, "session", emsg, ecode);
+		pThis->pEngine->onErr(pThis->pUsr, objinfo, emsg, ecode);
 	}
 }
 
@@ -276,8 +276,9 @@ relpSessRcvData(relpSess_t *pThis)
 		 */
 		pThis->sessState = eRelpSessState_BROKEN;
 		ABORT_FINALIZE(RELP_RET_SESSION_BROKEN);
-	} else if ((int) lenBuf == -1) { /* I don't know why we need to cast to int, but we must... */
+	} else if ((int) lenBuf == -1) {
 		if(errno != EAGAIN) {
+			callOnErr(pThis, "error when receiving data, session broken", RELP_RET_SESSION_BROKEN);
 			pThis->pEngine->dbgprint("errno %d during relp session %p, session broken\n",
 				errno, (void*)pThis);
 			pThis->sessState = eRelpSessState_BROKEN;
@@ -318,6 +319,7 @@ relpSessSendResponse(relpSess_t *pThis, relpTxnr_t txnr, unsigned char *pData, s
 finalize_it:
 	if(iRet != RELP_RET_OK) {
 		if(iRet == RELP_RET_IO_ERR) {
+			callOnErr(pThis, "io error, session broken", RELP_RET_SESSION_BROKEN);
 			pThis->pEngine->dbgprint("relp session %p is broken, io error\n", (void*)pThis);
 			pThis->sessState = eRelpSessState_BROKEN;
 			}
@@ -565,7 +567,7 @@ relpSessWaitState(relpSess_t *const pThis, const relpSessState_t stateExpected,
 		pfd.fd = sock;
 		pfd.events = POLLIN;
 		pThis->pEngine->dbgprint("relpSessWaitRsp waiting for data on "
-			"fd %d, timeout %d\n", sock, timeout);
+			"fd %d, timeout %d, state expected %d\n", sock, timeout, stateExpected);
 		nfds = poll(&pfd, 1, timeout*1000);
 		if(nfds == -1) {
 			if(errno == EINTR) {
@@ -597,6 +599,8 @@ relpSessWaitState(relpSess_t *const pThis, const relpSessState_t stateExpected,
 		iRet == RELP_RET_SESSION_BROKEN ||
 		relpEngineShouldStop(pThis->pEngine)) {
 		/* the session is broken! */
+		callOnErr(pThis, "error waiting on required session state, session broken",
+			RELP_RET_SESSION_BROKEN);
 		pThis->sessState = eRelpSessState_BROKEN;
 	}
 
@@ -628,6 +632,7 @@ relpSessRawSendCommand(relpSess_t *pThis, unsigned char *pCmd, size_t lenCmd,
 
 	if(iRet == RELP_RET_IO_ERR) {
 		pThis->pEngine->dbgprint("relp session %p flagged as broken, IO error\n", (void*)pThis);
+		callOnErr(pThis, "io error in RawSendCommand, session broken", RELP_RET_SESSION_BROKEN);
 		pThis->sessState = eRelpSessState_BROKEN;
 		ABORT_FINALIZE(RELP_RET_SESSION_BROKEN);
 	}
@@ -791,6 +796,7 @@ relpSessCBrspOpen(relpSess_t *pThis, relpFrame_t *pFrame)
 			pEngine->dbgprint("ignoring unknown server offer '%s'\n", pOffer->szName);
 		}
 	}
+
 	relpSessSetSessState(pThis, eRelpSessState_INIT_RSP_RCVD);
 
 finalize_it:
@@ -816,8 +822,10 @@ relpSessCltConnChkOffers(relpSess_t *pThis)
 		ABORT_FINALIZE(RELP_RET_RQD_FEAT_MISSING);
 
 finalize_it:
-	if(iRet != RELP_RET_OK)
+	if(iRet != RELP_RET_OK) {
+		callOnErr(pThis, "error in CltConnChkOffers, session broken", RELP_RET_SESSION_BROKEN);
 		pThis->sessState = eRelpSessState_BROKEN;
+	}
 
 	LEAVE_RELPFUNC;
 }
@@ -876,7 +884,6 @@ relpSessConnect(relpSess_t *pThis, int protFamily, unsigned char *port, unsigned
 	CHKRet(relpOffersToString(pOffers, NULL, 0, &pszOffers, &lenOffers));
 	CHKRet(relpOffersDestruct(&pOffers));
 
-pThis->pEngine->dbgprint("sending open command\n");
 	CHKRet(relpSessRawSendCommand(pThis, (unsigned char*)"open", 4, pszOffers, lenOffers,
 				      relpSessCBrspOpen));
 	relpSessSetSessState(pThis, eRelpSessState_INIT_CMD_SENT);

src/scrsp.c

@@ -31,6 +31,7 @@
  * development.
  */
 #include "config.h"
+#include <stdio.h>
 #include <stdlib.h>
 #include <ctype.h>
 #include <assert.h>
@@ -106,6 +107,10 @@ BEGINcommand(S, Rsp)
 		}
 		CHKRet(relpSendbufDestruct(&pSendbuf));
 	} else {
+		if(pSess->pEngine->onErr != NULL) {
+			pSess->pEngine->onErr(pSess->pUsr, "rsp command",
+				"peer sent error response", RELP_RET_RSP_STATE_ERR);
+		}
 		iRet = RELP_RET_RSP_STATE_ERR;
 		/* TODO: add a hock to a logger function of the caller */
 		relpSendbufDestruct(&pSendbuf); /* don't set error code */

src/tcp.c

@@ -658,14 +658,17 @@ relpTcpDestruct(relpTcp_t **ppThis)
 	RELPOBJ_assert(pThis, Tcp);
 
 	if(pThis->sock != -1) {
+		shutdown(pThis->sock, SHUT_RDWR);
 		close(pThis->sock);
 		pThis->sock = -1;
 	}
 
 	if(pThis->socks != NULL) {
 		/* if we have some sockets at this stage, we need to close them */
-		for(i = 1 ; i <= pThis->socks[0] ; ++i)
+		for(i = 1 ; i <= pThis->socks[0] ; ++i) {
+			shutdown(pThis->socks[i], SHUT_RDWR);
 			close(pThis->socks[i]);
+		}
 		free(pThis->socks);
 	}
 
@@ -2860,6 +2863,7 @@ relpTcpSend(relpTcp_t *const pThis, relpOctet_t *pBuf, ssize_t *pLenBuf)
 #endif /* defined(ENABLE_TLS) | defined(ENABLE_TLS_OPENSSL */
 		written = send(pThis->sock, pBuf, *pLenBuf, 0);
 		const int errno_save = errno;
+		pThis->pEngine->dbgprint("relpTcpSend: send data: %.*s\n", (int) *pLenBuf, pBuf);
 		pThis->pEngine->dbgprint("relpTcpSend: sock %d, lenbuf %zd, send returned %d [errno %d]\n",
 			(int)pThis->sock, *pLenBuf, (int) written, errno_save);
 		if(written == -1) {

tests/Makefile.am

@@ -1,21 +1,22 @@
-check_PROGRAMS=receive send
+check_PROGRAMS=receive send chkseq
 
 receive_SOURCES=receive.c
-#receive_LDADD=-lrelp
 receive_LDADD=../src/.libs/librelp.la
 receive_CFLAGS=$(AM_CFLAGS) -I${top_srcdir}/src $(WARN_CFLAGS)
 
 send_SOURCES=send.c
-#send_LDADD=-lrelp
 send_LDADD=../src/.libs/librelp.la
 send_CFLAGS=$(AM_CFLAGS) -I${top_srcdir}/src $(WARN_CFLAGS)
 
+chkseq_SOURCES=chkseq.c
+
 VALGRIND_TESTS= \
 	tls-basic-vg.sh \
 	duplicate-receiver-vg.sh
 
 TESTS=  basic.sh \
 	basic-realistic.sh \
+	receiver-abort.sh \
 	tls-basic.sh \
 	tls-basic-anon.sh \
 	tls-basic-certvalid.sh \
@@ -46,6 +47,7 @@ endif
 EXTRA_DIST=$(TESTS) \
 	$(VALGRIND_TESTS) \
 	dummyclient.py \
+	dummyserver.py \
 	test-framework.sh \
 	receive.c \
 	send.c \

tests/basic-realistic.sh

@@ -4,7 +4,7 @@
 # of messages
 . ${srcdir:=$(pwd)}/test-framework.sh
 NUMMESSAGES=50000
-startup_receiver $OPT_VERBOSE
+startup_receiver
 ./send -t 127.0.0.1 -p $TESTPORT -n$NUMMESSAGES $OPT_VERBOSE
 stop_receiver
 check_msg_count