1 files changed, 43 insertions, 12 deletions

diff --git a/plugins/omtesting/omtesting.c b/plugins/omtesting/omtesting.c
index c9f1e06..0df3063 100644
--- a/plugins/omtesting/omtesting.c
+++ b/plugins/omtesting/omtesting.c
@@ -22,7 +22,7 @@
  * NOTE: read comments in module-template.h to understand how this file
  *       works!
  *
- * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH.
+ * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -49,6 +49,7 @@
 #include <string.h>
 #include <ctype.h>
 #include <assert.h>
+#include <pthread.h>
 #include "dirty.h"
 #include "syslogd-types.h"
 #include "module-template.h"
@@ -63,7 +64,6 @@ MODULE_CNFNAME("omtesting")
  */
 DEF_OMOD_STATIC_DATA
 
-
 typedef struct _instanceData {
 	enum { MD_SLEEP, MD_FAIL, MD_RANDFAIL, MD_ALWAYS_SUSPEND }
 		mode;
@@ -74,8 +74,16 @@ typedef struct _instanceData {
 	int	iFailFrequency;
 	int	iResumeAfter;
 	int	iCurrRetries;
+	int	bFailed;	/* indicates if we are already in failed state - this is necessary
+	 			 * to work properly together with multiple worker instances.
+				 */
+	pthread_mutex_t mut;
 } instanceData;
 
+typedef struct wrkrInstanceData {
+	instanceData *pData;
+} wrkrInstanceData_t;
+
 typedef struct configSettings_s {
 	int bEchoStdout;	/* echo non-failed messages to stdout */
 } configSettings_t;
@@ -90,9 +98,15 @@ BEGINcreateInstance
 CODESTARTcreateInstance
 	pData->iWaitSeconds = 1;
 	pData->iWaitUSeconds = 0;
+	pthread_mutex_init(&pData->mut, NULL);
 ENDcreateInstance
 
 
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+ENDcreateWrkrInstance
+
+
 BEGINdbgPrintInstInfo
 CODESTARTdbgPrintInstInfo
 	dbgprintf("Action delays rule by %d second(s) and %d millisecond(s)\n",
@@ -115,6 +129,7 @@ static rsRetVal doFailOnResume(instanceData *pData)
 	dbgprintf("fail retry curr %d, max %d\n", pData->iCurrRetries, pData->iResumeAfter);
 	if(++pData->iCurrRetries == pData->iResumeAfter) {
 		iRet = RS_RET_OK;
+		pData->bFailed = 0;
 	} else {
 		iRet = RS_RET_SUSPENDED;
 	}
@@ -128,12 +143,18 @@ static rsRetVal doFail(instanceData *pData)
 {
 	DEFiRet;
 
-	dbgprintf("fail curr %d, frquency %d\n", pData->iCurrCallNbr, pData->iFailFrequency);
-	if(pData->iCurrCallNbr++ % pData->iFailFrequency == 0) {
-		pData->iCurrRetries = 0;
-		iRet = RS_RET_SUSPENDED;
+	dbgprintf("fail curr %d, frequency %d, bFailed %d\n", pData->iCurrCallNbr,
+		  pData->iFailFrequency, pData->bFailed);
+	if(pData->bFailed) {
+		ABORT_FINALIZE(RS_RET_SUSPENDED);
+	} else {
+		if(pData->iCurrCallNbr++ % pData->iFailFrequency == 0) {
+			pData->iCurrRetries = 0;
+			pData->bFailed = 1;
+			iRet = RS_RET_SUSPENDED;
+		}
 	}
-
+finalize_it:
 	RETiRet;
 }
 
@@ -170,11 +191,12 @@ static rsRetVal doRandFail(void)
 BEGINtryResume
 CODESTARTtryResume
 	dbgprintf("omtesting tryResume() called\n");
-	switch(pData->mode) {
+	pthread_mutex_lock(&pWrkrData->pData->mut);
+	switch(pWrkrData->pData->mode) {
 		case MD_SLEEP:
 			break;
 		case MD_FAIL:
-			iRet = doFailOnResume(pData);
+			iRet = doFailOnResume(pWrkrData->pData);
 			break;
 		case MD_RANDFAIL:
 			iRet = doRandFail();
@@ -182,13 +204,17 @@ CODESTARTtryResume
 		case MD_ALWAYS_SUSPEND:
 			iRet = RS_RET_SUSPENDED;
 	}
+	pthread_mutex_unlock(&pWrkrData->pData->mut);
 	dbgprintf("omtesting tryResume() returns iRet %d\n", iRet);
 ENDtryResume
 
 
 BEGINdoAction
+	instanceData *pData;
 CODESTARTdoAction
 	dbgprintf("omtesting received msg '%s'\n", ppString[0]);
+	pData = pWrkrData->pData;
+	pthread_mutex_lock(&pData->mut);
 	switch(pData->mode) {
 		case MD_SLEEP:
 			iRet = doSleep(pData);
@@ -208,18 +234,22 @@ CODESTARTdoAction
 		fprintf(stdout, "%s", ppString[0]);
 		fflush(stdout);
 	}
+	pthread_mutex_unlock(&pData->mut);
 	dbgprintf(":omtesting: end doAction(), iRet %d\n", iRet);
 ENDdoAction
 
 
 BEGINfreeInstance
 CODESTARTfreeInstance
-	/* we do not have instance data, so we do not need to
-	 * do anything here. -- rgerhards, 2007-07-25
-	 */
+	pthread_mutex_destroy(&pData->mut);
 ENDfreeInstance
 
 
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+ENDfreeWrkrInstance
+
+
 BEGINparseSelectorAct
 	int i;
 	uchar szBuf[1024];
@@ -313,6 +343,7 @@ ENDmodExit
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
 CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
 CODEqueryEtryPt_STD_CONF2_CNFNAME_QUERIES 
 ENDqueryEtryPt