milestone: calling sequence changed so that wti is passed to many functions

7 files changed, 71 insertions, 66 deletions

diff --git a/action.c b/action.c
index 5f49b44c..9ea3aa65 100644
--- a/action.c
+++ b/action.c
@@ -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 */
 
diff --git a/action.h b/action.h
index 17b22942..0724dc78 100644
--- a/action.h
+++ b/action.h
@@ -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;