api: add session lifecycle callbacks and tests

src/copen.c

@@ -158,6 +158,9 @@ BEGINcommand(S, Init)
 	CHKRet(relpSessSendResponse(pSess, pFrame->txnr, pszSrvOffers, lenSrvOffers));
 
 	pSess->bServerConnOpen = 1;
+	if(pSess->pEngine->onSessOpen != NULL) {
+		pSess->pEngine->onSessOpen(pSess->pUsr, pSess);
+	}
 
 finalize_it:
 	if(pszSrvOffers != NULL)
@@ -168,6 +171,9 @@ BEGINcommand(S, Init)
 		relpOffersDestruct(&pSrvOffers);
 
 	if(iRet != RELP_RET_OK) {
+		if(pSess->pEngine->onSessOpenFail != NULL) {
+			pSess->pEngine->onSessOpenFail(pSess->pUsr, pSess, iRet);
+		}
 		if(iRet == RELP_RET_RQD_FEAT_MISSING) {
 			strncpy(szErrMsg, "500 required command not supported by client", sizeof(szErrMsg));
 			lenErrMsg = 44;

src/librelp.h

@@ -220,6 +220,62 @@ relpRetVal relpEngineSetOnErr(relpEngine_t *pThis,
 				void (*pCB)(void*pUsr, char *objinfo, char*errmsg, relpRetVal errcode) );
 relpRetVal relpEngineSetOnGenericErr(relpEngine_t *pThis,
 				void (*pCB)(char *objinfo, char*errmsg, relpRetVal errcode) );
+/**
+ * Set a session-open callback.
+ *
+ * Callback signature:
+ *   void cb(void *pUsr, const relpSess_t *pSess)
+ *
+ * The callback is invoked after a server-side RELP session successfully
+ * completes the open handshake and is ready for traffic. The callback runs
+ * on the relpEngineRun() thread. The relpSess_t pointer is only valid for the
+ * duration of the callback and must not be retained.
+ *
+ * Callback parameters:
+ *
+ * pUsr  - user pointer from relpSrvSetUsrPtr()
+ * pSess - session being opened (read-only, ephemeral)
+ */
+relpRetVal relpEngineSetOnSessOpen(relpEngine_t *pThis,
+				void (*pCB)(void*pUsr, const relpSess_t *pSess) );
+/**
+ * Set a session-close callback.
+ *
+ * Callback signature:
+ *   void cb(void *pUsr, const relpSess_t *pSess, relpRetVal reason)
+ *
+ * The callback fires when a session is removed from the engine, regardless
+ * of the close cause (protocol close, I/O error, or engine shutdown). The
+ * reason parameter contains the relpRetVal that triggered teardown; engine
+ * shutdown uses RELP_RET_SESSION_CLOSED. The callback runs on the
+ * relpEngineRun() thread and pSess is valid only during the callback.
+ *
+ * Callback parameters:
+ *
+ * pUsr   - user pointer from relpSrvSetUsrPtr()
+ * pSess  - session being closed (read-only, ephemeral)
+ * reason - relpRetVal that caused teardown
+ */
+relpRetVal relpEngineSetOnSessClose(relpEngine_t *pThis,
+				void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) );
+/**
+ * Set a session-open-failed callback.
+ *
+ * Callback signature:
+ *   void cb(void *pUsr, const relpSess_t *pSess, relpRetVal reason)
+ *
+ * The callback fires when the server-side open handshake fails. It is
+ * invoked before the error response is sent. The callback runs on the
+ * relpEngineRun() thread and pSess is valid only during the callback.
+ *
+ * Callback parameters:
+ *
+ * pUsr   - user pointer from relpSrvSetUsrPtr()
+ * pSess  - session that failed to open (read-only, ephemeral)
+ * reason - relpRetVal describing the failure
+ */
+relpRetVal relpEngineSetOnSessOpenFail(relpEngine_t *pThis,
+				void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) );
 
 /* exposed server property set functions */
 relpRetVal relpSrvSetLstnPort(relpSrv_t *pThis, unsigned char *pLstnPort);

src/relp.c

@@ -220,11 +220,12 @@ relpEngineAddToSess(relpEngine_t *const pThis, relpSess_t *pSess)
  * rgerhards, 2008-03-17
  */
 static relpRetVal
-relpEngineDelSess(relpEngine_t *const pThis, relpEngSessLst_t *pSessLstEntry)
+relpEngineDelSess(relpEngine_t *const pThis, relpEngSessLst_t *pSessLstEntry, relpRetVal reason)
 {
 	ENTER_RELPFUNC;
 	RELPOBJ_assert(pThis, Engine);
 	assert(pSessLstEntry != NULL);
+	relpSess_t *const pSess = pSessLstEntry->pSess;
 
 #	if defined(HAVE_EPOLL_CREATE1) || defined(HAVE_EPOLL_CREATE)
 	delSessFromEpoll(pThis, pSessLstEntry);
@@ -234,6 +235,10 @@ relpEngineDelSess(relpEngine_t *const pThis, relpEngSessLst_t *pSessLstEntry)
 	--pThis->lenSessLst;
 	pthread_mutex_unlock(&pThis->mutSessLst);
 
+	if(pThis->onSessClose != NULL) {
+		pThis->onSessClose(pSess->pUsr, pSess, reason);
+	}
+
 	relpSessDestruct(&pSessLstEntry->pSess);
 	free(pSessLstEntry);
 
@@ -296,8 +301,7 @@ relpEngineDestruct(relpEngine_t **ppThis)
 	/* now destruct all currently existing sessions */
 	for(pSessL = pThis->pSessLstRoot ; pSessL != NULL ; pSessL = pSessLNxt) {
 		pSessLNxt = pSessL->pNext;
-		relpSessDestruct(&pSessL->pSess);
-		free(pSessL);
+		relpEngineDelSess(pThis, pSessL, RELP_RET_SESSION_CLOSED);
 	}
 
 	/* and now all existing servers... */
@@ -559,6 +563,35 @@ relpEngineSetOnGenericErr(relpEngine_t *const pThis, void (*pCB)(char *objinfo,
 	LEAVE_RELPFUNC;
 }
 
+relpRetVal PART_OF_API
+relpEngineSetOnSessOpen(relpEngine_t *const pThis, void (*pCB)(void*pUsr, const relpSess_t *pSess) )
+{
+	ENTER_RELPFUNC;
+	RELPOBJ_assert(pThis, Engine);
+	pThis->onSessOpen = pCB;
+	LEAVE_RELPFUNC;
+}
+
+relpRetVal PART_OF_API
+relpEngineSetOnSessClose(relpEngine_t *const pThis,
+		void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) )
+{
+	ENTER_RELPFUNC;
+	RELPOBJ_assert(pThis, Engine);
+	pThis->onSessClose = pCB;
+	LEAVE_RELPFUNC;
+}
+
+relpRetVal PART_OF_API
+relpEngineSetOnSessOpenFail(relpEngine_t *const pThis,
+		void (*pCB)(void*pUsr, const relpSess_t *pSess, relpRetVal reason) )
+{
+	ENTER_RELPFUNC;
+	RELPOBJ_assert(pThis, Engine);
+	pThis->onSessOpenFail = pCB;
+	LEAVE_RELPFUNC;
+}
+
 /* Deprecated, use relpEngineListnerConstruct() family of functions.
  * See there for further information.
  */
@@ -665,7 +698,7 @@ doRecv(relpEngine_t *const pThis, relpEngSessLst_t *pSessEtry, int sock)
 	if(localRet != RELP_RET_OK) {
 		pThis->dbgprint((char*)"relp session %d iRet %d, tearing it down\n",
 				sock, localRet);
-		relpEngineDelSess(pThis, pSessEtry);
+		relpEngineDelSess(pThis, pSessEtry, localRet);
 	}
 	return localRet;
 }
@@ -684,7 +717,7 @@ doSend(relpEngine_t *const pThis, relpEngSessLst_t *pSessEtry, int sock)
 	if(localRet != RELP_RET_OK) {
 		pThis->dbgprint((char*)"relp session %d iRet %d during send, tearing it down\n",
 				sock, localRet);
-		relpEngineDelSess(pThis, pSessEtry);
+		relpEngineDelSess(pThis, pSessEtry, localRet);
 	}
 }
 
@@ -801,7 +834,7 @@ handleSessIO(relpEngine_t *const pThis, epolld_t *epd)
 				if(localRet != RELP_RET_OK) {
 					pThis->dbgprint((char*)"relp session %d handshake iRet %d, tearing it down\n",
 							epd->sock, localRet);
-					relpEngineDelSess(pThis, pSessEtry);
+				    relpEngineDelSess(pThis, pSessEtry, localRet);
 				}
 #			else
 					pThis->dbgprint((char*)"librelp error: handshake retry requested in "
@@ -1001,7 +1034,7 @@ engineEventLoopRun(relpEngine_t *const pThis)
 								pThis->dbgprint((char*)"relp session %d handshake "
 										"iRet %d, tearing it down\n",
 										sock, localRet);
-								relpEngineDelSess(pThis, pSessEtry);
+							    relpEngineDelSess(pThis, pSessEtry, localRet);
 							}
 #						else
 							pThis->dbgprint((char*)"librelp error: handshake retry "

src/relp.h

@@ -130,11 +130,14 @@ struct relpEngine_s {
                                   unsigned char *pMsg, size_t lenMsg); /**< callback for "syslog" cmd */
         relpRetVal (*onSyslogRcv2)(void*, unsigned char*pHostname, unsigned char *pIP,
                                   unsigned char *pMsg, size_t lenMsg); /**< callback for "syslog" cmd */
-       relpRetVal (*onSyslogRcv3)(void*, unsigned char*pHostname, unsigned char *pIP,
+        relpRetVal (*onSyslogRcv3)(void*, unsigned char*pHostname, unsigned char *pIP,
                                   unsigned char *pPort, unsigned char *pMsg, size_t lenMsg); /**< callback for "syslog" cmd */
         void (*onAuthErr)(void*pUsr, char *authinfo, char*errmsg, relpRetVal errcode);
         void (*onErr)(void*pUsr, char *objinfo, char*errmsg, relpRetVal errcode);
         void (*onGenericErr)(char *objinfo, char*errmsg, relpRetVal errcode);
+        void (*onSessOpen)(void*pUsr, const relpSess_t *pSess);
+        void (*onSessClose)(void*pUsr, const relpSess_t *pSess, relpRetVal reason);
+        void (*onSessOpenFail)(void*pUsr, const relpSess_t *pSess, relpRetVal reason);
 	int protocolVersion; /**< version of the relp protocol supported by this engine */
 
 	/* Flags */

tests/Makefile.am

@@ -49,6 +49,7 @@ TESTS=  selftest_receive_watchdog.sh \
 	selftest_receive_usage.sh \
 	basic.sh \
 	basic-realistic.sh \
+	session-callbacks.sh \
 	receiver-abort.sh \
 	long-msg.sh \
 	oversize-msg-abort-errmsg.sh \
@@ -78,6 +79,8 @@ EXTRA_DIST=$(TESTS) \
 	set-envvars.in \
 	dummyclient.py \
 	dummyserver.py \
+	invalid-open.py \
+	session-callbacks.sh \
 	test-framework.sh \
 	receive.c \
 	send.c \
@@ -95,4 +98,3 @@ EXTRA_DIST=$(TESTS) \
 	tls-certs/ossl-server-cert.pem \
 	tls-certs/ossl-server-key.pem \
 	tls-certs/ossl-server-certchain.pem
-

tests/invalid-open.py

@@ -0,0 +1,24 @@
+#!/usr/bin/env python3
+
+import os
+import socket
+import time
+
+
+def main() -> None:
+    port = int(os.environ["TESTPORT"])
+    frame = b"1 open 12 relp_version\n"
+    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    sock.settimeout(2)
+    sock.connect(("127.0.0.1", port))
+    sock.sendall(frame)
+    try:
+        sock.recv(1024)
+    except socket.timeout:
+        pass
+    time.sleep(0.1)
+    sock.close()
+
+
+if __name__ == "__main__":
+    main()

tests/receive.c

@@ -42,6 +42,7 @@
 
 static FILE *errFile = NULL;
 static FILE *outFile = NULL;
+static FILE *sessFile = NULL;
 static char *pidFileName = NULL;
 
 static int immediate_exit = 0; /* if set to 1, force-exit as soon as possible */
@@ -199,6 +200,48 @@ onAuthErr(LIBRELP_ATTR_UNUSED void *pUsr, char *authinfo,
 	}
 }
 
+static void
+onSessOpen(void *pUsr, LIBRELP_ATTR_UNUSED const relpSess_t *pSess)
+{
+	if(sessFile == NULL) {
+		return;
+	}
+	if(pUsr != NULL && pUsr != userdata) {
+		fprintf(sessFile, "receive: session opened with unexpected pUsr %p\n", pUsr);
+	} else {
+		fprintf(sessFile, "receive: session opened\n");
+	}
+	fflush(sessFile);
+}
+
+static void
+onSessClose(void *pUsr, LIBRELP_ATTR_UNUSED const relpSess_t *pSess, relpRetVal reason)
+{
+	if(sessFile == NULL) {
+		return;
+	}
+	if(pUsr != NULL && pUsr != userdata) {
+		fprintf(sessFile, "receive: session closed reason=%d unexpected pUsr %p\n", reason, pUsr);
+	} else {
+		fprintf(sessFile, "receive: session closed reason=%d\n", reason);
+	}
+	fflush(sessFile);
+}
+
+static void
+onSessOpenFail(void *pUsr, LIBRELP_ATTR_UNUSED const relpSess_t *pSess, relpRetVal reason)
+{
+	if(sessFile == NULL) {
+		return;
+	}
+	if(pUsr != NULL && pUsr != userdata) {
+		fprintf(sessFile, "receive: session open failed reason=%d unexpected pUsr %p\n", reason, pUsr);
+	} else {
+		fprintf(sessFile, "receive: session open failed reason=%d\n", reason);
+	}
+	fflush(sessFile);
+}
+
 static void
 exit_hdlr(void)
 {
@@ -213,6 +256,9 @@ exit_hdlr(void)
 	if(outFile != NULL) {
 		fclose(outFile);
 	}
+	if(sessFile != NULL) {
+		fclose(sessFile);
+	}
 	if(pidFileName != NULL) {
 		unlink(pidFileName);
 	}
@@ -245,7 +291,7 @@ int main(int argc, char *argv[]) {
 	const char* outfile_name = NULL;
 
 #if defined(_AIX)
-	while((c = getopt(argc, argv, "a:c:Ae:F:l:m:No:O:P:p:TvW:x:y:z:")) != EOF) {
+	while((c = getopt(argc, argv, "a:c:Ae:F:l:m:No:O:P:p:TvW:C:x:y:z:")) != EOF) {
 #else
 	static struct option long_options[] =
 	{
@@ -256,6 +302,7 @@ int main(int argc, char *argv[]) {
 		{"authmode", required_argument, 0, 'a'},
 		{"pidfile", required_argument, 0, 'F'},
 		{"errorfile", required_argument, 0, 'e'},
+		{"sessfile", required_argument, 0, 'C'},
 		{"outfile", required_argument, 0, 'O'},
 		{"append-outfile", no_argument, 0, 'A'},
 		{"tls-lib", required_argument, 0, 'l'},
@@ -265,7 +312,7 @@ int main(int argc, char *argv[]) {
 		{0, 0, 0, 0}
 	};
 
-	while((c = getopt_long(argc, argv, "a:c:Ae:F:l:m:No:O:P:p:TvW:x:y:z:", long_options, &option_index)) != -1) {
+	while((c = getopt_long(argc, argv, "a:c:Ae:F:l:m:No:O:P:p:TvW:C:x:y:z:", long_options, &option_index)) != -1) {
 #endif
 		switch(c) {
 		case 'a':
@@ -277,6 +324,13 @@ int main(int argc, char *argv[]) {
 		case 'c':
 			tlsConfigCmd = optarg;
 			break;
+		case 'C':
+			if((sessFile = fopen((char*) optarg, "w")) == NULL) {
+				perror(optarg);
+				fprintf(stderr, "receive: error opening session callback file\n");
+				exit(1);
+			}
+			break;
 		case 'e':
 			if((errFile = fopen((char*) optarg, "w")) == NULL) {
 				perror(optarg);
@@ -417,6 +471,9 @@ int main(int argc, char *argv[]) {
 	TRY(relpEngineSetOnErr(pRelpEngine, onErr));
 	TRY(relpEngineSetOnGenericErr(pRelpEngine, onGenericErr));
 	TRY(relpEngineSetOnAuthErr(pRelpEngine, onAuthErr));
+	TRY(relpEngineSetOnSessOpen(pRelpEngine, onSessOpen));
+	TRY(relpEngineSetOnSessClose(pRelpEngine, onSessClose));
+	TRY(relpEngineSetOnSessOpenFail(pRelpEngine, onSessOpenFail));
 
 	if(tlslib != NULL) {
 		TRY(relpEngineSetTLSLibByName(pRelpEngine, tlslib));

tests/session-callbacks.sh

@@ -0,0 +1,18 @@
+#!/bin/bash
+. ${srcdir:=$(pwd)}/test-framework.sh
+
+CALLBACK_FILE=${TESTDIR}/session-callbacks.log
+
+startup_receiver -C ${CALLBACK_FILE} $OPT_VERBOSE
+
+./send -t 127.0.0.1 -p $TESTPORT -m "testmessage" $OPT_VERBOSE
+
+$PYTHON ${srcdir}/invalid-open.py
+
+stop_receiver
+
+check_output "testmessage"
+check_output "session opened" ${CALLBACK_FILE}
+check_output "session closed reason=" ${CALLBACK_FILE}
+check_output "session open failed reason=" ${CALLBACK_FILE}
+terminate