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