diff options
| -rw-r--r-- | msg.c | 3 | ||||
| -rw-r--r-- | msg.h | 1 | ||||
| -rw-r--r-- | queue.c | 31 | ||||
| -rw-r--r-- | queue.h | 1 | ||||
| -rw-r--r-- | syslogd.c | 6 |
5 files changed, 29 insertions, 13 deletions
@@ -286,10 +286,9 @@ msg_t* MsgDup(msg_t* pOld) * is a so slow operation that recration of the caches does not count. * rgerhards, 2008-01-03 */ -rsRetVal MsgSerialize(uchar **ppOutBuf, size_t *pLenBuf, void *pUsr) +static rsRetVal MsgSerialize(msg_t *pThis, uchar **ppOutBuf, size_t *pLenBuf) { DEFiRet; - msg_t* pThis = pUsr; rsCStrObj *pCStr; dbgprintf("MsgSerialize in\n"); @@ -109,7 +109,6 @@ typedef struct msg msg_t; /* new name */ PROTOTYPEObjClassInit(Msg); char* getProgramName(msg_t*); msg_t* MsgConstruct(void); -rsRetVal MsgSerialize(uchar **ppOutBuf, size_t *pLenBuf, void *pUsr); rsRetVal MsgDestruct(msg_t * pM); msg_t* MsgDup(msg_t* pOld); msg_t *MsgAddRef(msg_t *pM); @@ -235,7 +235,7 @@ rsRetVal qAddDisk(queue_t *pThis, void* pUsr) assert(pThis != NULL); dbgprintf("writing to file %d\n", pThis->tVars.disk.fd); - CHKiRet((objSerialize(pUsr))(pBuf, &lenBuf, pUsr)); // TODO: hier weiter machen! + CHKiRet((objSerialize(pUsr))(pUsr, pBuf, &lenBuf)); // TODO: hier weiter machen! i = write(pThis->tVars.disk.fd, "entry\n", 6); dbgprintf("write wrote %d bytes, errno: %d, err %s\n", i, errno, strerror(errno)); @@ -360,12 +360,15 @@ queueWorker(void *arg) pthread_exit(0); } -/* Constructor for the queue object */ +/* Constructor for the queue object + * This constructs the data structure, but does not yet start the queue. That + * is done by queueStart(). The reason is that we want to give the caller a chance + * to modify some parameters before the queue is actually started. + */ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iMaxQueueSize, rsRetVal (*pConsumer)(void*)) { DEFiRet; queue_t *pThis; - int i; assert(ppThis != NULL); assert(pConsumer != NULL); @@ -411,12 +414,6 @@ rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iMaxQueueSize, /* call type-specific constructor */ CHKiRet(pThis->qConstruct(pThis)); - /* now fire up the worker thread */ - pThis->bDoRun = 1; /* we are NOT done (else worker would immediately terminate) */ - i = pthread_create(&pThis->thrdWorker, NULL, queueWorker, (void*) pThis); - dbgprintf("Worker thread for queue 0x%lx, type %d started with state %d.\n", - (unsigned long) pThis, (int) qType, i); - finalize_it: if(iRet == RS_RET_OK) { *ppThis = pThis; @@ -429,6 +426,22 @@ finalize_it: } +/* start up the queue - it must have been constructed and parameters defined + * before. + */ +rsRetVal queueStart(queue_t *pThis) +{ + int i; + + /* fire up the worker thread */ + pThis->bDoRun = 1; /* we are NOT done (else worker would immediately terminate) */ + i = pthread_create(&pThis->thrdWorker, NULL, queueWorker, (void*) pThis); + dbgprintf("Worker thread for queue 0x%lx, type %d started with state %d.\n", + (unsigned long) pThis, (int) pThis->qType, i); + + return RS_RET_OK; +} + /* destructor for the queue object */ rsRetVal queueDestruct(queue_t *pThis) { @@ -84,5 +84,6 @@ typedef struct queue_s { rsRetVal queueDestruct(queue_t *pThis); rsRetVal queueEnqObj(queue_t *pThis, void *pUsr); rsRetVal queueConstruct(queue_t **ppThis, queueType_t qType, int iMaxQueueSize, rsRetVal (*pConsumer)(void*)); +rsRetVal queueStart(queue_t *pThis); #endif /* #ifndef QUEUE_H_INCLUDED */ @@ -2268,7 +2268,6 @@ static int parseLegacySyslogMsg(msg_t *pMsg, int flags) void logmsg(int pri, msg_t *pMsg, int flags) { - DEFiRet; char *msg; char PRItext[20]; @@ -3356,6 +3355,11 @@ init(void) fprintf(stderr, "fatal error %d: could not create message queue - rsyslogd can not run!\n", iRet); exit(1); } + CHKiRet_Hdlr(queueStart(pMsgQueue)) { + /* no queue is fatal, we need to give up in that case... */ + fprintf(stderr, "fatal error %d: could not start message queue - rsyslogd can not run!\n", iRet); + exit(1); + } Initialized = 1; |