1 files changed, 54 insertions, 12 deletions

diff --git a/tcps_sess.c b/tcps_sess.c
index d0edc018..2a3ec0df 100644
--- a/tcps_sess.c
+++ b/tcps_sess.c
@@ -43,6 +43,7 @@
 #include "errmsg.h"
 #include "netstrm.h"
 #include "msg.h"
+#include "datetime.h"
 
 
 /* static data */
@@ -50,6 +51,7 @@ DEFobjStaticHelpers
 DEFobjCurrIf(glbl)
 DEFobjCurrIf(errmsg)
 DEFobjCurrIf(netstrm)
+DEFobjCurrIf(datetime)
 
 static int iMaxLine; /* maximum size of a single message */
 
@@ -191,6 +193,52 @@ SetUsrP(tcps_sess_t *pThis, void *pUsr)
 }
 
 
+/* 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
+doSubmitMessage(tcps_sess_t *pThis)
+{
+	msg_t *pMsg;
+	struct syslogTime stTime;
+	time_t ttGenTime;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, tcps_sess);
+	
+	//TODO: if((iTimeRequery == 0) || (iNbrTimeUsed++ % iTimeRequery) == 0) {
+		datetime.getCurrTime(&stTime, &ttGenTime);
+	//}
+	/* we now create our own message object and submit it to the queue */
+	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);
+	pMsg->iLenRawMsg = pThis->iMsg;
+	MsgSetInputName(pMsg, (char*)pThis->pSrv->pszInputName);
+	MsgSetFlowControlType(pMsg, eFLOWCTL_LIGHT_DELAY);
+	pMsg->msgFlags  = NEEDS_PARSING | PARSE_HOSTNAME;
+	pMsg->bParseHOSTNAME = 1;
+	MsgSetRcvFrom(pMsg, (char*)pThis->fromHost);
+	CHKiRet(MsgSetRcvFromIP(pMsg, pThis->fromHostIP));
+	CHKiRet(submitMsg(pMsg));
+
+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
@@ -230,9 +278,7 @@ 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,
-				      PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, pThis->pSrv->pszInputName, NULL, 0);
-		pThis->bAtStrtOfFram = 1;
+		doSubmitMessage(pThis);
 	}
 
 finalize_it:
@@ -313,9 +359,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,
-				      	      PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, pThis->pSrv->pszInputName, NULL, 0);
-			pThis->iMsg = 0;
+			doSubmitMessage(pThis);
 			/* 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...
@@ -326,9 +370,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? */
-			parseAndSubmitMessage(pThis->fromHost, pThis->fromHostIP, pThis->pMsg, pThis->iMsg,
-				      	      PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, pThis->pSrv->pszInputName, NULL, 0);
-			pThis->iMsg = 0;
+			doSubmitMessage(pThis);
 			pThis->inputState = eAtStrtFram;
 		} else {
 			/* IMPORTANT: here we copy the actual frame content to the message - for BOTH framing modes!
@@ -345,9 +387,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,
-				      	      PARSE_HOSTNAME, eFLOWCTL_LIGHT_DELAY, pThis->pSrv->pszInputName, NULL, 0);
-				pThis->iMsg = 0;
+				doSubmitMessage(pThis);
 				pThis->inputState = eAtStrtFram;
 			}
 		}
@@ -432,6 +472,7 @@ CODESTARTObjClassExit(tcps_sess)
 	/* release objects we no longer need */
 	objRelease(errmsg, CORE_COMPONENT);
 	objRelease(netstrm, LM_NETSTRMS_FILENAME);
+	objRelease(datetime, CORE_COMPONENT);
 ENDObjClassExit(tcps_sess)
 
 
@@ -443,6 +484,7 @@ 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(glbl, CORE_COMPONENT));
 	iMaxLine = glbl.GetMaxLine(); /* get maximum size we currently support */