Merge branch 'v7-stable-call-ruleset-queue' into v7-stable

Conflicts:
	ChangeLog

5 files changed, 47 insertions, 4 deletions

diff --git a/ChangeLog b/ChangeLog
index fd4e3942..97ea6f5a 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,6 @@
+Version 7.4.7  [v7.4-stable] 2013-11-??
+- bugfix: call to ruleset with async queue did not use the queue
+  closes: http://bugzilla.adiscon.com/show_bug.cgi?id=443
 ---------------------------------------------------------------------------
 Version 7.4.6  [v7.4-stable] 2013-10-31
 - bugfix: potential abort during HUP
diff --git a/grammar/rainerscript.c b/grammar/rainerscript.c
index b03e7c12..a1de6442 100644
--- a/grammar/rainerscript.c
+++ b/grammar/rainerscript.c
@@ -2279,7 +2279,8 @@ cnfstmtPrintOnly(struct cnfstmt *stmt, int indent, sbool subtree)
 		break;
 	case S_CALL:
 		cstr = es_str2cstr(stmt->d.s_call.name, NULL);
-		doIndent(indent); dbgprintf("CALL [%s]\n", cstr);
+		doIndent(indent); dbgprintf("CALL [%s, queue:%d]\n", cstr,
+			stmt->d.s_call.ruleset == NULL ? 0 : 1);
 		free(cstr);
 		break;
 	case S_ACT:
@@ -3110,8 +3111,14 @@ cnfstmtOptimizeCall(struct cnfstmt *stmt)
 		stmt->nodetype = S_NOP;
 		goto done;
 	}
-	DBGPRINTF("CALL obtained ruleset ptr %p for ruleset %s\n", pRuleset, rsName);
-	stmt->d.s_call.stmt = pRuleset->root;
+	DBGPRINTF("CALL obtained ruleset ptr %p for ruleset %s [hasQueue:%d]\n",
+		  pRuleset, rsName, rulesetHasQueue(pRuleset));
+	if(rulesetHasQueue(pRuleset)) {
+		stmt->d.s_call.ruleset = pRuleset;
+	} else {
+		stmt->d.s_call.ruleset = NULL;
+		stmt->d.s_call.stmt = pRuleset->root;
+	}
 done:
 	free(rsName);
 	return;
diff --git a/grammar/rainerscript.h b/grammar/rainerscript.h
index d00cc4c3..06573307 100644
--- a/grammar/rainerscript.h
+++ b/grammar/rainerscript.h
@@ -5,6 +5,7 @@
 #include <typedefs.h>
 #include <sys/types.h>
 #include <regex.h>
+#include "typedefs.h"
 
 
 #define	LOG_NFACILITIES	24	/* current number of syslog facilities */
@@ -164,6 +165,7 @@ struct cnfstmt {
 		struct {
 			es_str_t *name;
 			struct cnfstmt *stmt;
+			ruleset_t *ruleset;	/* non-NULL if the ruleset has a queue assigned */
 		} s_call;
 		struct {
 			uchar pmask[LOG_NFACILITIES+1];	/* priority mask */
diff --git a/runtime/ruleset.c b/runtime/ruleset.c
index 5bf7ac03..1afb4039 100644
--- a/runtime/ruleset.c
+++ b/runtime/ruleset.c
@@ -284,6 +284,30 @@ execStop(batch_t *pBatch, sbool *active)
 	}
 	RETiRet;
 }
+static rsRetVal
+execCall(struct cnfstmt *stmt, batch_t *pBatch, sbool *active)
+{
+	msg_t *pMsg;
+	int i;
+	DEFiRet;
+	if(stmt->d.s_call.ruleset == NULL) {
+		scriptExec(stmt->d.s_call.stmt, pBatch, active);
+	} else {
+		for(i = 0 ; i < batchNumMsgs(pBatch) ; ++i) {
+			CHKmalloc(pMsg = MsgDup((msg_t*) pBatch->pElem[i].pMsg));
+			DBGPRINTF("CALL: forwarding message %d to async ruleset %p\n",
+				  i, stmt->d.s_call.ruleset->pQueue);
+			MsgSetFlowControlType(pMsg, eFLOWCTL_NO_DELAY);
+			MsgSetRuleset(pMsg, stmt->d.s_call.ruleset);
+			/* Note: we intentionally use submitMsg2() here, as we process messages
+			 * that were already run through the rate-limiter.
+			 */
+			submitMsg2(pMsg);
+		}
+	}
+finalize_it:
+	RETiRet;
+}
 
 /* for details, see scriptExec() header comment! */
 // save current filter, evaluate new one
@@ -535,7 +559,7 @@ 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);
+			execCall(stmt, pBatch, active);
 			break;
 		case S_IF:
 			execIf(stmt, pBatch, active);
diff --git a/runtime/ruleset.h b/runtime/ruleset.h
index cbf8243b..64fe92ff 100644
--- a/runtime/ruleset.h
+++ b/runtime/ruleset.h
@@ -90,6 +90,13 @@ rulesetGetName(ruleset_t *pRuleset)
 	return pRuleset->pszName;
 }
 
+/* returns 1 if the ruleset has a queue associtated, 0 if not */
+static inline int
+rulesetHasQueue(ruleset_t *pRuleset)
+{
+	return pRuleset->pQueue == NULL ? 0 : 1;
+}
+
 
 /* we will most probably convert this module back to traditional C
  * calling sequence, so here we go...