diff options
author | Rainer Gerhards <rgerhards@adiscon.com> | 2009-07-30 11:09:15 +0200 |
---|---|---|
committer | Rainer Gerhards <rgerhards@adiscon.com> | 2009-07-30 11:09:15 +0200 |
commit | a6bda9b93f21cdbec1d7312078535eb092f32cb0 (patch) | |
tree | e1b62c000eea2b81e0e67694b2d8129e252f830c /action.c | |
parent | feeb622c4e0c622559df803f8df6da39bf3015e7 (diff) | |
download | rsyslog-a6bda9b93f21cdbec1d7312078535eb092f32cb0.tar.gz rsyslog-a6bda9b93f21cdbec1d7312078535eb092f32cb0.tar.xz rsyslog-a6bda9b93f21cdbec1d7312078535eb092f32cb0.zip |
bugfix: discard action did not work (did not discard messages)
Diffstat (limited to 'action.c')
-rw-r--r-- | action.c | 56 |
1 files changed, 37 insertions, 19 deletions
@@ -808,41 +808,50 @@ tryDoAction(action_t *pAction, batch_t *pBatch, int *pnElem) assert(pBatch != NULL); assert(pnElem != NULL); +dbgprintf("XXXX: ENTER tryDoAction elt 0 state %d\n", pBatch->pElem[0].state); i = pBatch->iDoneUpTo; /* all messages below that index are processed */ iElemProcessed = 0; iCommittedUpTo = i; while(iElemProcessed <= *pnElem && i < pBatch->nElem) { pMsg = (msg_t*) pBatch->pElem[i].pUsrp; - dbgprintf("submitBatch: i:%d, batch size %d, to process %d, pMsg: %p\n", i, pBatch->nElem, *pnElem, pMsg);//remove later! - localRet = actionProcessMessage(pAction, pMsg); - dbgprintf("action call returned %d\n", localRet); - if(localRet == RS_RET_OK) { - /* mark messages as committed */ - while(iCommittedUpTo < i) { - pBatch->pElem[iCommittedUpTo++].state = BATCH_STATE_COMM; - } - } else if(localRet == RS_RET_PREVIOUS_COMMITTED) { - /* mark messages as committed */ - while(iCommittedUpTo < i - 1) { - pBatch->pElem[iCommittedUpTo++].state = BATCH_STATE_COMM; + dbgprintf("submitBatch: i:%d, batch size %d, to process %d, pMsg: %p, state %d\n", i, pBatch->nElem, *pnElem, pMsg, pBatch->pElem[i].state);//remove later! + if(pBatch->pElem[i].state != BATCH_STATE_DISC) { + localRet = actionProcessMessage(pAction, pMsg); + dbgprintf("action call returned %d\n", localRet); + if(localRet == RS_RET_OK) { + /* mark messages as committed */ + while(iCommittedUpTo < i) { + pBatch->pElem[iCommittedUpTo++].state = BATCH_STATE_COMM; + } + } else if(localRet == RS_RET_PREVIOUS_COMMITTED) { + /* mark messages as committed */ + while(iCommittedUpTo < i - 1) { + pBatch->pElem[iCommittedUpTo++].state = BATCH_STATE_COMM; + } + pBatch->pElem[i].state = BATCH_STATE_SUB; + } else if(localRet == RS_RET_PREVIOUS_COMMITTED) { + pBatch->pElem[i].state = BATCH_STATE_SUB; + } else if(localRet == RS_RET_DISCARDMSG) { + pBatch->pElem[i].state = BATCH_STATE_DISC; +dbgprintf("XXXX: discardmsg! change state to _DISC: %d\n", pBatch->pElem[i].state); + } else { + iRet = localRet; + FINALIZE; } - pBatch->pElem[i].state = BATCH_STATE_SUB; - } else if(localRet == RS_RET_PREVIOUS_COMMITTED) { - pBatch->pElem[i].state = BATCH_STATE_SUB; - } else { - iRet = localRet; - FINALIZE; } ++i; ++iElemProcessed; } finalize_it: - if(pBatch->iDoneUpTo != iCommittedUpTo) { + if(pBatch->nElem == 1 && pBatch->pElem[0].state == BATCH_STATE_DISC) { + iRet = RS_RET_DISCARDMSG; + } else if(pBatch->iDoneUpTo != iCommittedUpTo) { *pnElem += iCommittedUpTo - pBatch->iDoneUpTo; pBatch->iDoneUpTo = iCommittedUpTo; } +dbgprintf("XXXX: done tryDoAction elt 0 state %d, iret %d\n", pBatch->pElem[0].state, iRet); RETiRet; } @@ -865,6 +874,7 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) bDone = 0; do { localRet = tryDoAction(pAction, pBatch, &nElem); +dbgprintf("XXXX: submitBatch got state %d\n", localRet); if( localRet == RS_RET_OK || localRet == RS_RET_PREVIOUS_COMMITTED || localRet == RS_RET_DEFER_COMMIT) { @@ -874,10 +884,15 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) localRet = finishBatch(pAction); } +dbgprintf("XXXX: submitBatch got state %d\n", localRet); if( localRet == RS_RET_OK || localRet == RS_RET_PREVIOUS_COMMITTED || localRet == RS_RET_DEFER_COMMIT) { bDone = 1; + } else if(localRet == RS_RET_DISCARDMSG) { + iRet = RS_RET_DISCARDMSG; /* TODO: verify this sequence -- rgerhards, 2009-07-30 */ + bDone = 1; +dbgprintf("XXXX: submitBatch DONE state %d\n", localRet); } else if(localRet == RS_RET_SUSPENDED) { ; /* do nothing, this will retry the full batch */ } else if(localRet == RS_RET_ACTION_FAILED) { @@ -897,8 +912,10 @@ submitBatch(action_t *pAction, batch_t *pBatch, int nElem) bDone = 1; } } +dbgprintf("XXXX: submitBatch pre while state %d\n", localRet); } while(!bDone); /* do .. while()! */ +dbgprintf("XXXX: END submitBatch elt 0 state %d, iRet %d\n", pBatch->pElem[0].state, iRet); RETiRet; } @@ -1134,6 +1151,7 @@ actionWriteToAction(action_t *pAction) * So let's enqueue our message for execution. -- rgerhards, 2007-07-24 */ iRet = qqueueEnqObj(pAction->pQueue, pAction->f_pMsg->flowCtlType, (void*) MsgAddRef(pAction->f_pMsg)); +dbgprintf("XXXX: queueEnqObj returned %d\n", iRet); if(iRet == RS_RET_OK) pAction->f_prevcount = 0; /* message processed, so we start a new cycle */ |