added new testing module imdiag

which enables to talk to the rsyslog core at runtime. The current
implementation is only a beginning, but can be expanded over time

1 files changed, 21 insertions, 6 deletions

diff --git a/tcps_sess.c b/tcps_sess.c
index c564caea..c4548804 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -58,6 +58,7 @@ static int iMaxLine; /* maximum size of a single message */
 
 /* forward definitions */
 static rsRetVal Close(tcps_sess_t *pThis);
+static rsRetVal defaultDoSubmitMessage(tcps_sess_t *pThis, uchar*, int);
 
 
 /* Standard-Constructor */
@@ -65,6 +66,7 @@ BEGINobjConstruct(tcps_sess) /* be sure to specify the object type also in END m
 		pThis->iMsg = 0; /* just make sure... */
 		pThis->bAtStrtOfFram = 1; /* indicate frame header expected */
 		pThis->eFraming = TCP_FRAMING_OCTET_STUFFING; /* just make sure... */
+		pThis->DoSubmitMessage = defaultDoSubmitMessage;
 		/* now allocate the message reception buffer */
 		CHKmalloc(pThis->pMsg = (uchar*) malloc(sizeof(uchar) * iMaxLine + 1));
 finalize_it:
@@ -206,6 +208,15 @@ SetUsrP(tcps_sess_t *pThis, void *pUsr)
 }
 
 
+static rsRetVal
+SetOnMsgReceive(tcps_sess_t *pThis, rsRetVal (*OnMsgReceive)(tcps_sess_t*, uchar*, int))
+{
+	DEFiRet;
+	pThis->DoSubmitMessage = OnMsgReceive;
+	RETiRet;
+}
+
+
 /* This is a helper for submitting the message to the rsyslog core.
  * It does some common processing, including resetting the various
  * state variables to a "processed" state.
@@ -217,8 +228,11 @@ SetUsrP(tcps_sess_t *pThis, void *pUsr)
  * rgerhards, 2009-04-23
  */
 static rsRetVal
-doSubmitMessage(tcps_sess_t *pThis)
+defaultDoSubmitMessage(tcps_sess_t *pThis, uchar *pszMsg, int iLenMsg)
 {
+// TODO: make calling this overridable so that the diag module can ask to be called
+// and so it can do its work right in this entry point (but we need to check that
+// we have the capability to send a reply at this point).
 	msg_t *pMsg;
 	struct syslogTime stTime;
 	time_t ttGenTime;
@@ -233,7 +247,7 @@ doSubmitMessage(tcps_sess_t *pThis)
 	CHKiRet(msgConstructWithTime(&pMsg, &stTime, ttGenTime));
 	/* first trim the buffer to what we have actually received */
 	CHKmalloc(pMsg->pszRawMsg = malloc(sizeof(uchar) * pThis->iMsg));
-	memcpy(pMsg->pszRawMsg, pThis->pMsg, pThis->iMsg);
+	memcpy(pMsg->pszRawMsg, pszMsg, iLenMsg);
 	pMsg->iLenRawMsg = pThis->iMsg;
 	MsgSetInputName(pMsg, pThis->pLstnInfo->pszInputName);
 	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
@@ -291,7 +305,7 @@ PrepareClose(tcps_sess_t *pThis)
 		 * this case.
 		 */
 		dbgprintf("Extra data at end of stream in legacy syslog/tcp message - processing\n");
-		doSubmitMessage(pThis);
+		pThis->DoSubmitMessage(pThis, pThis->pMsg, pThis->iMsg);
 	}
 
 finalize_it:
@@ -372,7 +386,7 @@ processDataRcvd(tcps_sess_t *pThis, char c)
 		if(pThis->iMsg >= iMaxLine) {
 			/* emergency, we now need to flush, no matter if we are at end of message or not... */
 			dbgprintf("error: message received is larger than max msg size, we split it\n");
-			doSubmitMessage(pThis);
+			pThis->DoSubmitMessage(pThis, pThis->pMsg, pThis->iMsg);
 			/* we might think if it is better to ignore the rest of the
 			 * message than to treat it as a new one. Maybe this is a good
 			 * candidate for a configuration parameter...
@@ -383,7 +397,7 @@ processDataRcvd(tcps_sess_t *pThis, char c)
 		if((   (c == '\n')
 		   || ((pThis->pSrv->addtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER) && (c == pThis->pSrv->addtlFrameDelim))
 		   ) && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delimiter? */
-			doSubmitMessage(pThis);
+			pThis->DoSubmitMessage(pThis, pThis->pMsg, pThis->iMsg);
 			pThis->inputState = eAtStrtFram;
 		} else {
 			/* IMPORTANT: here we copy the actual frame content to the message - for BOTH framing modes!
@@ -400,7 +414,7 @@ processDataRcvd(tcps_sess_t *pThis, char c)
 			pThis->iOctetsRemain--;
 			if(pThis->iOctetsRemain < 1) {
 				/* we have end of frame! */
-				doSubmitMessage(pThis);
+				pThis->DoSubmitMessage(pThis, pThis->pMsg, pThis->iMsg);
 				pThis->inputState = eAtStrtFram;
 			}
 		}
@@ -474,6 +488,7 @@ CODESTARTobjQueryInterface(tcps_sess)
 	pIf->SetHostIP = SetHostIP;
 	pIf->SetStrm = SetStrm;
 	pIf->SetMsgIdx = SetMsgIdx;
+	pIf->SetOnMsgReceive = SetOnMsgReceive;
 finalize_it:
 ENDobjQueryInterface(tcps_sess)