summaryrefslogtreecommitdiffstats
path: root/tcps_sess.c
diff options
context:
space:
mode:
Diffstat (limited to 'tcps_sess.c')
-rw-r--r--tcps_sess.c180
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:
*/