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

4 files changed, 33 insertions, 31 deletions

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 */