1 files changed, 144 insertions, 36 deletions

diff --git a/tcps_sess.c b/tcps_sess.c
index 3024c313..ea9032b3 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -36,6 +36,7 @@
 
 #include "rsyslog.h"
 #include "dirty.h"
+#include "unicode-helper.h"
 #include "module-template.h"
 #include "net.h"
 #include "tcpsrv.h"
@@ -43,6 +44,9 @@
 #include "obj.h"
 #include "errmsg.h"
 #include "netstrm.h"
+#include "msg.h"
+#include "datetime.h"
+#include "prop.h"
 
 
 /* static data */
@@ -50,9 +54,12 @@ DEFobjStaticHelpers
 DEFobjCurrIf(glbl)
 DEFobjCurrIf(errmsg)
 DEFobjCurrIf(netstrm)
+DEFobjCurrIf(prop)
+DEFobjCurrIf(datetime)
 
 static int iMaxLine; /* maximum size of a single message */
 
+
 /* forward definitions */
 static rsRetVal Close(tcps_sess_t *pThis);
 
@@ -95,11 +102,10 @@ CODESTARTobjDestruct(tcps_sess)
 	}
 	/* now destruct our own properties */
 	if(pThis->fromHost != NULL)
-		free(pThis->fromHost);
+		CHKiRet(prop.Destruct(&pThis->fromHost));
 	if(pThis->fromHostIP != NULL)
-		free(pThis->fromHostIP);
-	if(pThis->pMsg != NULL)
-		free(pThis->pMsg);
+		CHKiRet(prop.Destruct(&pThis->fromHostIP));
+	free(pThis->pMsg);
 ENDobjDestruct(tcps_sess)
 
 
@@ -121,12 +127,13 @@ SetHost(tcps_sess_t *pThis, uchar *pszHost)
 
 	ISOBJ_TYPE_assert(pThis, tcps_sess);
 
-	if(pThis->fromHost != NULL) {
-		free(pThis->fromHost);
-	}
-	
-	pThis->fromHost = pszHost;
+	if(pThis->fromHost == NULL)
+		CHKiRet(prop.Construct(&pThis->fromHost));
 
+	CHKiRet(prop.SetString(pThis->fromHost, pszHost, ustrlen(pszHost)));
+
+finalize_it:
+	free(pszHost); /* we must free according to our (old) calling conventions */
 	RETiRet;
 }
 
@@ -141,12 +148,13 @@ SetHostIP(tcps_sess_t *pThis, uchar *pszHostIP)
 
 	ISOBJ_TYPE_assert(pThis, tcps_sess);
 
-	if(pThis->fromHostIP != NULL) {
-		free(pThis->fromHostIP);
-	}
-	
-	pThis->fromHostIP = pszHostIP;
+	if(pThis->fromHostIP == NULL)
+		CHKiRet(prop.Construct(&pThis->fromHostIP));
 
+	CHKiRet(prop.SetString(pThis->fromHostIP, pszHostIP, ustrlen(pszHostIP)));
+
+finalize_it:
+	free(pszHostIP);
 	RETiRet;
 }
 
@@ -182,6 +190,18 @@ SetTcpsrv(tcps_sess_t *pThis, tcpsrv_t *pSrv)
 }
 
 
+/* set our parent listener info*/
+static rsRetVal
+SetLstnInfo(tcps_sess_t *pThis, tcpLstnPortList_t *pLstnInfo)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, tcps_sess);
+	assert(pLstnInfo != NULL);
+	pThis->pLstnInfo = pLstnInfo;
+	RETiRet;
+}
+
+
 static rsRetVal
 SetUsrP(tcps_sess_t *pThis, void *pUsr)
 {
@@ -191,6 +211,73 @@ 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.
+ * Note that this function is also called if we had a buffer overflow
+ * due to a too-long message. So far, there is no indication this 
+ * happened and it may be worth thinking about different handling
+ * of this case (what obviously would require a change to this
+ * function or some related code).
+ * rgerhards, 2009-04-23
+ */
+static rsRetVal
+defaultDoSubmitMessage(tcps_sess_t *pThis, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub)
+{
+	msg_t *pMsg;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, tcps_sess);
+	
+	if(pThis->iMsg == 0) {
+		DBGPRINTF("discarding zero-sized message\n");
+		FINALIZE;
+	}
+
+	if(pThis->DoSubmitMessage != NULL) {
+		pThis->DoSubmitMessage(pThis, pThis->pMsg, pThis->iMsg);
+		FINALIZE;
+	}
+
+	/* we now create our own message object and submit it to the queue */
+	CHKiRet(msgConstructWithTime(&pMsg, stTime, ttGenTime));
+	MsgSetRawMsg(pMsg, (char*)pThis->pMsg, pThis->iMsg);
+	MsgSetInputName(pMsg, pThis->pLstnInfo->pInputName);
+	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
+	pMsg->msgFlags  = NEEDS_PARSING | PARSE_HOSTNAME;
+	pMsg->bParseHOSTNAME = 1;
+	MsgSetRcvFrom(pMsg, pThis->fromHost);
+	CHKiRet(MsgSetRcvFromIP(pMsg, pThis->fromHostIP));
+	MsgSetRuleset(pMsg, pThis->pLstnInfo->pRuleset);
+
+	if(pMultiSub == NULL) {
+		CHKiRet(submitMsg(pMsg));
+	} else {
+		pMultiSub->ppMsgs[pMultiSub->nElem++] = pMsg;
+		if(pMultiSub->nElem == pMultiSub->maxElem)
+			CHKiRet(multiSubmitMsg(pMultiSub));
+	}
+
+
+finalize_it:
+	/* reset status variables */
+	pThis->bAtStrtOfFram = 1;
+	pThis->iMsg = 0;
+
+	RETiRet;
+}
+
+
+
 /* This should be called before a normal (non forced) close
  * of a TCP session. This function checks if there is any unprocessed
  * message left in the TCP stream. Such a message is probably a
@@ -205,6 +292,8 @@ SetUsrP(tcps_sess_t *pThis, void *pUsr)
 static rsRetVal
 PrepareClose(tcps_sess_t *pThis)
 {
+	struct syslogTime stTime;
+	time_t ttGenTime;
 	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, tcps_sess);
@@ -230,9 +319,8 @@ PrepareClose(tcps_sess_t *pThis)
 		 * this case.
 		 */
 		dbgprintf("Extra data at end of stream in legacy syslog/tcp message - processing\n");
-		parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg, MSG_PARSE_HOSTNAME,
-				      NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */
-		pThis->bAtStrtOfFram = 1;
+		datetime.getCurrTime(&stTime, &ttGenTime);
+		defaultDoSubmitMessage(pThis, &stTime, ttGenTime, NULL);
 	}
 
 finalize_it:
@@ -251,10 +339,11 @@ Close(tcps_sess_t *pThis)
 
 	ISOBJ_TYPE_assert(pThis, tcps_sess);
 	netstrm.Destruct(&pThis->pStrm);
-	free(pThis->fromHost);
-	pThis->fromHost = NULL; /* not really needed, but... */
-	free(pThis->fromHostIP);
-	pThis->fromHostIP = NULL; /* not really needed, but... */
+	if(pThis->fromHost != NULL) {
+		prop.Destruct(&pThis->fromHost);
+	}
+	if(pThis->fromHostIP != NULL)
+		prop.Destruct(&pThis->fromHostIP);
 
 	RETiRet;
 }
@@ -267,7 +356,7 @@ Close(tcps_sess_t *pThis)
  * rgerhards, 2008-03-14
  */
 static rsRetVal
-processDataRcvd(tcps_sess_t *pThis, char c)
+processDataRcvd(tcps_sess_t *pThis, char c, struct syslogTime *stTime, time_t ttGenTime, multi_submit_t *pMultiSub)
 {
 	DEFiRet;
 	ISOBJ_TYPE_assert(pThis, tcps_sess);
@@ -313,9 +402,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");
-			parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg,
-					      MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */
-			pThis->iMsg = 0;
+			defaultDoSubmitMessage(pThis, stTime, ttGenTime, pMultiSub);
 			/* 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...
@@ -323,10 +410,10 @@ processDataRcvd(tcps_sess_t *pThis, char c)
 			 */
 		}
 
-		if(c == '\n' && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delemiter? */
-			parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg,
-					      MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */
-			pThis->iMsg = 0;
+		if((   (c == '\n')
+		   || ((pThis->pSrv->addtlFrameDelim != TCPSRV_NO_ADDTL_DELIMITER) && (c == pThis->pSrv->addtlFrameDelim))
+		   ) && pThis->eFraming == TCP_FRAMING_OCTET_STUFFING) { /* record delimiter? */
+			defaultDoSubmitMessage(pThis, stTime, ttGenTime, pMultiSub);
 			pThis->inputState = eAtStrtFram;
 		} else {
 			/* IMPORTANT: here we copy the actual frame content to the message - for BOTH framing modes!
@@ -343,9 +430,7 @@ processDataRcvd(tcps_sess_t *pThis, char c)
 			pThis->iOctetsRemain--;
 			if(pThis->iOctetsRemain < 1) {
 				/* we have end of frame! */
-				parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg,
-					      MSG_PARSE_HOSTNAME, NOFLAG, eFLOWCTL_LIGHT_DELAY, NULL); /* TODO: add real InputName */
-				pThis->iMsg = 0;
+				defaultDoSubmitMessage(pThis, stTime, ttGenTime, pMultiSub);
 				pThis->inputState = eAtStrtFram;
 			}
 		}
@@ -366,27 +451,46 @@ processDataRcvd(tcps_sess_t *pThis, char c)
  * RS_RET_OK, which means the session should be kept open
  * or anything else, which means it must be closed.
  * rgerhards, 2008-03-01
+ * As a performance optimization, we pick up the timestamp here. Acutally,
+ * this *is* the *correct* reception step for all the data we received, because
+ * we have just received a bunch of data! -- rgerhards, 2009-06-16
  */
+#define NUM_MULTISUB 1024
 static rsRetVal
 DataRcvd(tcps_sess_t *pThis, char *pData, size_t iLen)
 {
-	DEFiRet;
+	multi_submit_t multiSub;
+	msg_t *pMsgs[NUM_MULTISUB];
+	struct syslogTime stTime;
+	time_t ttGenTime;
 	char *pEnd;
+	DEFiRet;
 
 	ISOBJ_TYPE_assert(pThis, tcps_sess);
 	assert(pData != NULL);
 	assert(iLen > 0);
 
+	datetime.getCurrTime(&stTime, &ttGenTime);
+	multiSub.ppMsgs = pMsgs;
+	multiSub.maxElem = NUM_MULTISUB;
+	multiSub.nElem = 0;
+
 	 /* We now copy the message to the session buffer. */
 	pEnd = pData + iLen; /* this is one off, which is intensional */
 
 	while(pData < pEnd) {
-		CHKiRet(processDataRcvd(pThis, *pData++));
+		CHKiRet(processDataRcvd(pThis, *pData++, &stTime, ttGenTime, &multiSub));
+	}
+
+	if(multiSub.nElem > 0) {
+		/* submit anything that was not yet submitted */
+		CHKiRet(multiSubmitMsg(&multiSub));
 	}
 
 finalize_it:
 	RETiRet;
 }
+#undef NUM_MULTISUB
 
 
 /* queryInterface function
@@ -414,10 +518,12 @@ CODESTARTobjQueryInterface(tcps_sess)
 
 	pIf->SetUsrP = SetUsrP;
 	pIf->SetTcpsrv = SetTcpsrv;
+	pIf->SetLstnInfo = SetLstnInfo;
 	pIf->SetHost = SetHost;
 	pIf->SetHostIP = SetHostIP;
 	pIf->SetStrm = SetStrm;
 	pIf->SetMsgIdx = SetMsgIdx;
+	pIf->SetOnMsgReceive = SetOnMsgReceive;
 finalize_it:
 ENDobjQueryInterface(tcps_sess)
 
@@ -430,6 +536,8 @@ CODESTARTObjClassExit(tcps_sess)
 	/* release objects we no longer need */
 	objRelease(errmsg, CORE_COMPONENT);
 	objRelease(netstrm, LM_NETSTRMS_FILENAME);
+	objRelease(datetime, CORE_COMPONENT);
+	objRelease(prop, CORE_COMPONENT);
 ENDObjClassExit(tcps_sess)
 
 
@@ -441,6 +549,8 @@ BEGINObjClassInit(tcps_sess, 1, OBJ_IS_CORE_MODULE) /* class, version - CHANGE c
 	/* request objects we use */
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));
 	CHKiRet(objUse(netstrm, LM_NETSTRMS_FILENAME));
+	CHKiRet(objUse(datetime, CORE_COMPONENT));
+	CHKiRet(objUse(prop, CORE_COMPONENT));
 
 	CHKiRet(objUse(glbl, CORE_COMPONENT));
 	iMaxLine = glbl.GetMaxLine(); /* get maximum size we currently support */
@@ -451,7 +561,5 @@ BEGINObjClassInit(tcps_sess, 1, OBJ_IS_CORE_MODULE) /* class, version - CHANGE c
 	OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, tcps_sessConstructFinalize);
 ENDObjClassInit(tcps_sess)
 
-
-
 /* vim:set ai:
  */