diff options
| -rw-r--r-- | action.c | 65 | ||||
| -rw-r--r-- | action.h | 4 | ||||
| -rw-r--r-- | runtime/queue.c | 9 | ||||
| -rw-r--r-- | runtime/queue.h | 4 | ||||
| -rw-r--r-- | runtime/ruleset.c | 46 | ||||
| -rw-r--r-- | runtime/ruleset.h | 5 | ||||
| -rw-r--r-- | tools/syslogd.c | 4 |
7 files changed, 71 insertions, 66 deletions
@@ -119,10 +119,10 @@ #define NO_TIME_PROVIDED 0 /* indicate we do not provide any cached time */ /* forward definitions */ -static rsRetVal processBatchMain(action_t *pAction, batch_t *pBatch, int*); -static rsRetVal doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch); -static rsRetVal doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch); -static rsRetVal doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch); +static rsRetVal processBatchMain(void *pVoid, batch_t *pBatch, wti_t *pWti, int*); +static rsRetVal doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti); +static rsRetVal doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti); +static rsRetVal doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti); /* object static data (once for all instances) */ /* TODO: make this an object! DEFobjStaticHelpers -- rgerhards, 2008-03-05 */ @@ -428,7 +428,7 @@ actionConstructFinalize(action_t *pThis, struct nvlst *lst) * spec. -- rgerhards, 2008-01-30 */ CHKiRet(qqueueConstruct(&pThis->pQueue, cs.ActionQueType, 1, cs.iActionQueueSize, - (rsRetVal (*)(void*, batch_t*, int*))processBatchMain)); + processBatchMain)); obj.SetName((obj_t*) pThis->pQueue, pszAName); qqueueSetpAction(pThis->pQueue, pThis); @@ -905,7 +905,7 @@ done: RETiRet; * rgerhards, 2008-01-28 */ rsRetVal -actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams) +actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams, wti_t *pWti) { DEFiRet; @@ -916,6 +916,8 @@ actionCallDoAction(action_t *pThis, msg_t *pMsg, void *actParams) getActStateName(pThis), pThis->iActionNbr); pThis->bHadAutoCommit = 0; +dbgprintf("DDDD: calling action id %d for wti %p\n", pThis->iActionNbr, pWti); +#warning TODO: add action instance check here iRet = pThis->pMod->mod.om.doAction(actParams, pMsg->msgFlags, pThis->pModData); switch(iRet) { case RS_RET_OK: @@ -954,7 +956,7 @@ finalize_it: * rgerhards, 2008-01-28 */ static inline rsRetVal -actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutdownImmediate) +actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutdownImmediate, wti_t *pWti) { DEFiRet; @@ -965,7 +967,7 @@ actionProcessMessage(action_t *pThis, msg_t *pMsg, void *actParams, int *pbShutd if(pThis->pMod->mod.om.SetShutdownImmdtPtr != NULL) pThis->pMod->mod.om.SetShutdownImmdtPtr(pThis->pModData, pbShutdownImmediate); if(pThis->eState == ACT_STATE_ITX) - CHKiRet(actionCallDoAction(pThis, pMsg, actParams)); + CHKiRet(actionCallDoAction(pThis, pMsg, actParams, pWti)); iRet = getReturnCode(pThis); finalize_it: @@ -1035,7 +1037,7 @@ finalize_it: * rgerhards, 2009-05-12 */ static inline rsRetVal -tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) +tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem, wti_t *pWti) { int i; int iElemProcessed; @@ -1060,7 +1062,7 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) if(batchIsValidElem(pBatch, i)) { pMsg = pBatch->pElem[i].pMsg; localRet = actionProcessMessage(pAction, pMsg, pBatch->pElem[i].staticActParams, - pBatch->pbShutdownImmediate); + pBatch->pbShutdownImmediate, pWti); DBGPRINTF("action %p call returned %d\n", pAction, localRet); /* Note: we directly modify the batch object state, because we know that * wo do not overwrite BATCH_STATE_DISC indicators! @@ -1112,7 +1114,7 @@ finalize_it: * rgerhards, 2009-05-12 */ static rsRetVal -submitBatch(action_t *pAction, batch_t *pBatch, int nElem) +submitBatch(action_t *pAction, batch_t *pBatch, int nElem, wti_t *pWti) { int i; int bDone; @@ -1126,7 +1128,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) wasDoneTo = pBatch->iDoneUpTo; bDone = 0; do { - localRet = tryDoAction(pAction, pBatch, &nElem); + localRet = tryDoAction(pAction, pBatch, &nElem, pWti); if(localRet == RS_RET_FORCE_TERM) { ABORT_FINALIZE(RS_RET_FORCE_TERM); } @@ -1165,8 +1167,8 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) /* retry with half as much. Depth is log_2 batchsize, so recursion is not too deep */ DBGPRINTF("submitBatch recursing trying to find and exclude the culprit " "for iRet %d\n", localRet); - submitBatch(pAction, pBatch, nElem / 2); - submitBatch(pAction, pBatch, nElem - (nElem / 2)); + submitBatch(pAction, pBatch, nElem / 2, pWti); + submitBatch(pAction, pBatch, nElem - (nElem / 2), pWti); bDone = 1; } } @@ -1246,12 +1248,12 @@ prepareBatch(action_t *pAction, batch_t *pBatch, sbool **activeSave, int *bMustR * rgerhards, 2009-05-12 */ static inline rsRetVal -processAction(action_t *pAction, batch_t *pBatch) +processAction(action_t *pAction, batch_t *pBatch, wti_t* pWti) { DEFiRet; assert(pBatch != NULL); - CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem)); + CHKiRet(submitBatch(pAction, pBatch, pBatch->nElem, pWti)); iRet = finishBatch(pAction, pBatch); finalize_it: @@ -1265,12 +1267,13 @@ finalize_it: * rgerhards, 2009-04-22 */ static rsRetVal -processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) +processBatchMain(void *pVoid, batch_t *pBatch, wti_t *pWti, int *pbShutdownImmediate) { int *pbShutdownImmdtSave; sbool *activeSave; int bMustRestoreActivePtr = 0; rsRetVal localRet; + action_t *pAction = (action_t*) pVoid; DEFiRet; assert(pBatch != NULL); @@ -1289,7 +1292,7 @@ processBatchMain(action_t *pAction, batch_t *pBatch, int *pbShutdownImmediate) d_pthread_mutex_lock(&pAction->mutActExec); pthread_cleanup_push(mutexCancelCleanup, &pAction->mutActExec); - iRet = processAction(pAction, pBatch); + iRet = processAction(pAction, pBatch, pWti); pthread_cleanup_pop(1); /* unlock mutex */ @@ -1382,7 +1385,7 @@ static rsRetVal setActionQueType(void __attribute__((unused)) *pVal, uchar *pszT * rgerhards, 2010-06-08 */ static inline rsRetVal -doSubmitToActionQ(action_t *pAction, msg_t *pMsg) +doSubmitToActionQ(action_t *pAction, msg_t *pMsg, wti_t *pWti) { DEFiRet; @@ -1410,7 +1413,7 @@ finalize_it: * be filtered out before calling us (what is done currently!). */ rsRetVal -actionWriteToAction(action_t *pAction, msg_t *pMsg) +actionWriteToAction(action_t *pAction, msg_t *pMsg, wti_t *pWti) { DEFiRet; @@ -1465,7 +1468,7 @@ actionWriteToAction(action_t *pAction, msg_t *pMsg) /* When we reach this point, we have a valid, non-disabled action. * So let's enqueue our message for execution. -- rgerhards, 2007-07-24 */ - iRet = doSubmitToActionQ(pAction, pMsg); + iRet = doSubmitToActionQ(pAction, pMsg, pWti); finalize_it: RETiRet; @@ -1476,7 +1479,7 @@ finalize_it: * pthread_cleanup_push() POSIX macro... */ static inline rsRetVal -doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) +doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch, wti_t *pWti) { msg_t *pMsg; DEFiRet; @@ -1491,7 +1494,7 @@ doActionCallAction(action_t *pAction, batch_t *pBatch, int idxBtch) } /* call the output driver */ - iRet = actionWriteToAction(pAction, pMsg); + iRet = actionWriteToAction(pAction, pMsg, pWti); finalize_it: /* we need to update the batch to handle failover processing correctly */ @@ -1551,7 +1554,7 @@ activateActions(void) * rgerhards, 2010-06-08 */ static rsRetVal -doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) +doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { time_t now = 0; time_t lastAct; @@ -1589,7 +1592,7 @@ doSubmitToActionQNotAllMarkBatch(action_t *pAction, batch_t *pBatch) } } - iRet = doSubmitToActionQBatch(pAction, pBatch); + iRet = doSubmitToActionQBatch(pAction, pBatch, pWti); free(pBatch->active); pBatch->active = activeSave; @@ -1667,7 +1670,7 @@ doQueueEnqObjDirectBatch(action_t *pAction, batch_t *pBatch) * rgerhards, 2010-06-08 */ static rsRetVal -doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) +doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { int i; DEFiRet; @@ -1685,7 +1688,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); if( batchIsValidElem(pBatch, i) && (pAction->bExecWhenPrevSusp == 0 || pBatch->pElem[i].bPrevWasSuspended == 1)) { - doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg); + doSubmitToActionQ(pAction, pBatch->pElem[i].pMsg, pWti); } } } @@ -1700,7 +1703,7 @@ doSubmitToActionQBatch(action_t *pAction, batch_t *pBatch) * rgerhards, 2010-06-23 */ static inline rsRetVal -helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) +helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { int i; DEFiRet; @@ -1713,7 +1716,7 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) pAction->bExecWhenPrevSusp, pBatch->pElem[i].bPrevWasSuspended); if( batchIsValidElem(pBatch, i) && ((pAction->bExecWhenPrevSusp == 0) || pBatch->pElem[i].bPrevWasSuspended) ) { - doActionCallAction(pAction, pBatch, i); + doActionCallAction(pAction, pBatch, i, pWti); } } @@ -1725,13 +1728,13 @@ helperSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) */ #pragma GCC diagnostic ignored "-Wempty-body" static rsRetVal -doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch) +doSubmitToActionQComplexBatch(action_t *pAction, batch_t *pBatch, wti_t *pWti) { DEFiRet; d_pthread_mutex_lock(&pAction->mutAction); pthread_cleanup_push(mutexCancelCleanup, &pAction->mutAction); - iRet = helperSubmitToActionQComplexBatch(pAction, pBatch); + iRet = helperSubmitToActionQComplexBatch(pAction, pBatch, pWti); d_pthread_mutex_unlock(&pAction->mutAction); pthread_cleanup_pop(0); /* remove mutex cleanup handler */ @@ -69,7 +69,7 @@ struct action_s { struct modInfo_s *pMod;/* pointer to output module handling this selector */ void *pModData; /* pointer to module data - content is module-specific */ sbool bRepMsgHasMsg; /* "message repeated..." has msg fragment in it (0-no, 1-yes) */ - rsRetVal (*submitToActQ)(action_t *, batch_t *);/* function submit message to action queue */ + rsRetVal (*submitToActQ)(action_t *, batch_t *, wti_t*);/* function submit message to action queue */ rsRetVal (*qConstruct)(struct queue_s *pThis); enum { ACT_STRING_PASSING = 0, ACT_ARRAY_PASSING = 1, ACT_MSG_PASSING = 2, ACT_JSON_PASSING = 3} @@ -97,7 +97,7 @@ rsRetVal actionDestruct(action_t *pThis); rsRetVal actionDbgPrint(action_t *pThis); rsRetVal actionSetGlobalResumeInterval(int iNewVal); rsRetVal actionDoAction(action_t *pAction); -rsRetVal actionWriteToAction(action_t *pAction, msg_t *pMsg); +rsRetVal actionWriteToAction(action_t *pAction, msg_t *pMsg, wti_t*); rsRetVal actionCallHUPHdlr(action_t *pAction); rsRetVal actionClassInit(void); rsRetVal addAction(action_t **ppAction, modInfo_t *pMod, void *pModData, omodStringRequest_t *pOMSR, struct cnfparamvals *actParams, struct nvlst *lst, int bSuspended); diff --git a/runtime/queue.c b/runtime/queue.c index 66cb7218..92ac2425 100644 --- a/runtime/queue.c +++ b/runtime/queue.c @@ -986,7 +986,7 @@ static rsRetVal qAddDirect(qqueue_t *pThis, msg_t* pMsg) singleBatch.pElem = &batchObj; singleBatch.eltState = &batchState; singleBatch.active = &active; - iRet = pThis->pConsumer(pThis->pAction, &singleBatch, &pThis->bShutdownImmediate); + iRet = pThis->pConsumer(pThis->pAction, &singleBatch, NULL, &pThis->bShutdownImmediate); /* delete the batch string params: TODO: create its own "class" for this */ for(i = 0 ; i < CONF_OMOD_NUMSTRINGS_MAXSIZE ; ++i) { free(batchObj.staticActStrings[i]); @@ -1013,7 +1013,8 @@ rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch) * We use our knowledge about the batch_t structure below, but without that, we * pay a too-large performance toll... -- rgerhards, 2009-04-22 */ - iRet = pThis->pConsumer(pThis->pAction, pBatch, NULL); +#warning TODO: handle wti ptr! + iRet = pThis->pConsumer(pThis->pAction, pBatch, NULL, NULL); RETiRet; } @@ -1319,7 +1320,7 @@ finalize_it: * to modify some parameters before the queue is actually started. */ rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, - int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*,int*)) + int iMaxQueueSize, rsRetVal (*pConsumer)(void*, batch_t*, wti_t*, int*)) { DEFiRet; qqueue_t *pThis; @@ -1878,7 +1879,7 @@ ConsumerReg(qqueue_t *pThis, wti_t *pWti) pthread_setcancelstate(PTHREAD_CANCEL_ENABLE, &iCancelStateSave); - CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, &pThis->bShutdownImmediate)); + CHKiRet(pThis->pConsumer(pThis->pAction, &pWti->batch, pWti, &pThis->bShutdownImmediate)); /* we now need to check if we should deliberately delay processing a bit * and, if so, do that. -- rgerhards, 2008-01-30 diff --git a/runtime/queue.h b/runtime/queue.h index 844523ad..01b4f351 100644 --- a/runtime/queue.h +++ b/runtime/queue.h @@ -103,7 +103,7 @@ struct queue_s { * the user really wanted...). -- rgerhards, 2008-04-02 */ /* end dequeue time window */ - rsRetVal (*pConsumer)(void *,batch_t*,int*); /* user-supplied consumer function for dequeued messages */ + rsRetVal (*pConsumer)(void *,batch_t*, wti_t*,int*); /* user-supplied consumer function for dequeued messages */ /* calling interface for pConsumer: arg1 is the global user pointer from this structure, arg2 is the * user pointer array that was dequeued (actual sample: for actions, arg1 is the pAction and arg2 * is pointer to an array of message message pointers), arg3 is a pointer to an interger which is zero @@ -201,7 +201,7 @@ rsRetVal qqueueStart(qqueue_t *pThis); rsRetVal qqueueSetMaxFileSize(qqueue_t *pThis, size_t iMaxFileSize); rsRetVal qqueueSetFilePrefix(qqueue_t *pThis, uchar *pszPrefix, size_t iLenPrefix); rsRetVal qqueueConstruct(qqueue_t **ppThis, queueType_t qType, int iWorkerThreads, - int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, int*)); + int iMaxQueueSize, rsRetVal (*pConsumer)(void*,batch_t*, wti_t *, int*)); rsRetVal qqueueEnqObjDirectBatch(qqueue_t *pThis, batch_t *pBatch); int queueCnfParamsSet(struct nvlst *lst); rsRetVal qqueueApplyCnfParam(qqueue_t *pThis, struct nvlst *lst); diff --git a/runtime/ruleset.c b/runtime/ruleset.c index 15e9e5ea..3e5223f9 100644 --- a/runtime/ruleset.c +++ b/runtime/ruleset.c @@ -67,8 +67,8 @@ static struct cnfparamblk rspblk = }; /* forward definitions */ -static rsRetVal processBatch(batch_t *pBatch); -static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active); +static rsRetVal processBatch(batch_t *pBatch, wti_t *pWti); +static rsRetVal scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti); /* ---------- linked-list key handling functions (ruleset) ---------- */ @@ -168,7 +168,7 @@ finalize_it: * rgerhards, 2010-06-15 */ static inline rsRetVal -processBatchMultiRuleset(batch_t *pBatch) +processBatchMultiRuleset(batch_t *pBatch, wti_t *pWti) { ruleset_t *currRuleset; batch_t snglRuleBatch; @@ -206,7 +206,7 @@ processBatchMultiRuleset(batch_t *pBatch) snglRuleBatch.nElem = iNew; /* was left just right by the for loop */ batchSetSingleRuleset(&snglRuleBatch, 1); /* process temp batch */ - processBatch(&snglRuleBatch); + processBatch(&snglRuleBatch, pWti); batchFree(&snglRuleBatch); } while(bHaveUnprocessed == 1); @@ -226,12 +226,12 @@ static inline void freeActive(sbool *active) { free(active); } /* for details, see scriptExec() header comment! */ /* call action for all messages with filter on */ static rsRetVal -execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execAct(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { DEFiRet; dbgprintf("RRRR: execAct [%s]: batch of %d elements, active %p\n", modGetName(stmt->d.act->pMod), batchNumMsgs(pBatch), active); pBatch->active = active; - stmt->d.act->submitToActQ(stmt->d.act, pBatch); + stmt->d.act->submitToActQ(stmt->d.act, pBatch, pWti); RETiRet; } @@ -292,7 +292,7 @@ execStop(batch_t *pBatch, sbool *active) // set new filter, inverted // perform else (if any messages) static rsRetVal -execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { sbool *newAct; int i; @@ -321,7 +321,7 @@ execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) } if(stmt->d.s_if.t_then != NULL) { - scriptExec(stmt->d.s_if.t_then, pBatch, newAct); + scriptExec(stmt->d.s_if.t_then, pBatch, newAct, pWti); } if(stmt->d.s_if.t_else != NULL) { for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { @@ -331,7 +331,7 @@ execIf(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) && (active == NULL || active[i])) newAct[i] = !newAct[i]; } - scriptExec(stmt->d.s_if.t_else, pBatch, newAct); + scriptExec(stmt->d.s_if.t_else, pBatch, newAct, pWti); } freeActive(newAct); finalize_it: @@ -340,7 +340,7 @@ finalize_it: /* for details, see scriptExec() header comment! */ static void -execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { sbool *newAct; msg_t *pMsg; @@ -367,7 +367,7 @@ execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) } if(stmt->d.s_prifilt.t_then != NULL) { - scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct); + scriptExec(stmt->d.s_prifilt.t_then, pBatch, newAct, pWti); } if(stmt->d.s_prifilt.t_else != NULL) { for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) { @@ -377,7 +377,7 @@ execPRIFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) && (active == NULL || active[i])) newAct[i] = !newAct[i]; } - scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct); + scriptExec(stmt->d.s_prifilt.t_else, pBatch, newAct, pWti); } freeActive(newAct); } @@ -476,7 +476,7 @@ done: /* for details, see scriptExec() header comment! */ static void -execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) +execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active, wti_t *pWti) { sbool *thenAct; sbool bRet; @@ -495,7 +495,7 @@ execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) DBGPRINTF("batch: item %d PROPFILT %d\n", i, thenAct[i]); } - scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct); + scriptExec(stmt->d.s_propfilt.t_then, pBatch, thenAct, pWti); freeActive(thenAct); } @@ -510,7 +510,7 @@ execPROPFILT(struct cnfstmt *stmt, batch_t *pBatch, sbool *active) * rgerhards, 2012-09-04 */ static rsRetVal -scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) +scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active, wti_t *pWti) { DEFiRet; struct cnfstmt *stmt; @@ -528,7 +528,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) execStop(pBatch, active); break; case S_ACT: - execAct(stmt, pBatch, active); + execAct(stmt, pBatch, active, pWti); break; case S_SET: execSet(stmt, pBatch, active); @@ -537,16 +537,16 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) execUnset(stmt, pBatch, active); break; case S_CALL: - scriptExec(stmt->d.s_call.stmt, pBatch, active); + scriptExec(stmt->d.s_call.stmt, pBatch, active, pWti); break; case S_IF: - execIf(stmt, pBatch, active); + execIf(stmt, pBatch, active, pWti); break; case S_PRIFILT: - execPRIFILT(stmt, pBatch, active); + execPRIFILT(stmt, pBatch, active, pWti); break; case S_PROPFILT: - execPROPFILT(stmt, pBatch, active); + execPROPFILT(stmt, pBatch, active, pWti); break; default: dbgprintf("error: unknown stmt type %u during exec\n", @@ -565,7 +565,7 @@ scriptExec(struct cnfstmt *root, batch_t *pBatch, sbool *active) * rgerhards, 2005-10-13 */ static rsRetVal -processBatch(batch_t *pBatch) +processBatch(batch_t *pBatch, wti_t *pWti) { ruleset_t *pThis; DEFiRet; @@ -577,9 +577,9 @@ processBatch(batch_t *pBatch) if(pThis == NULL) pThis = ourConf->rulesets.pDflt; ISOBJ_TYPE_assert(pThis, ruleset); - CHKiRet(scriptExec(pThis->root, pBatch, NULL)); + CHKiRet(scriptExec(pThis->root, pBatch, NULL, pWti)); } else { - CHKiRet(processBatchMultiRuleset(pBatch)); + CHKiRet(processBatchMultiRuleset(pBatch, pWti)); } finalize_it: diff --git a/runtime/ruleset.h b/runtime/ruleset.h index 8bfd4920..05da144d 100644 --- a/runtime/ruleset.h +++ b/runtime/ruleset.h @@ -46,7 +46,7 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ rsRetVal (*Destruct)(ruleset_t **ppThis); rsRetVal (*DestructAllActions)(rsconf_t *conf); rsRetVal (*SetName)(ruleset_t *pThis, uchar *pszName); - rsRetVal (*ProcessBatch)(batch_t*); + rsRetVal (*ProcessBatch)(batch_t*, wti_t *); rsRetVal (*GetRuleset)(rsconf_t *conf, ruleset_t **ppThis, uchar*); rsRetVal (*SetDefaultRuleset)(rsconf_t *conf, uchar*); rsRetVal (*SetCurrRuleset)(rsconf_t *conf, uchar*); @@ -64,8 +64,9 @@ BEGINinterface(ruleset) /* name must also be changed in ENDinterface macro! */ /* AddRule() removed */ /*TODO:REMOVE*/rsRetVal (*IterateAllActions)(rsconf_t *conf, rsRetVal (*pFunc)(void*, void*), void* pParam); void (*AddScript)(ruleset_t *pThis, struct cnfstmt *script); + /* v8: changed processBatch interface */ ENDinterface(ruleset) -#define rulesetCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */ +#define rulesetCURR_IF_VERSION 8 /* increment whenever you change the interface structure! */ /* prototypes */ diff --git a/tools/syslogd.c b/tools/syslogd.c index aaeb9866..7597b05d 100644 --- a/tools/syslogd.c +++ b/tools/syslogd.c @@ -560,13 +560,13 @@ finalize_it: * for the main queue. */ static rsRetVal -msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, int *pbShutdownImmediate) +msgConsumer(void __attribute__((unused)) *notNeeded, batch_t *pBatch, wti_t *pWti, int *pbShutdownImmediate) { DEFiRet; assert(pBatch != NULL); pBatch->pbShutdownImmediate = pbShutdownImmediate; /* TODO: move this to batch creation! */ preprocessBatch(pBatch); - ruleset.ProcessBatch(pBatch); + ruleset.ProcessBatch(pBatch, pWti); //TODO: the BATCH_STATE_COMM must be set somewhere down the road, but we //do not have this yet and so we emulate -- 2010-06-10 int i; |