From 800ac1889b99057f1e6670d4ce5941bda33b6773 Mon Sep 17 00:00:00 2001
From: Rainer Gerhards <rgerhards@adiscon.com>
Date: Fri, 4 Jan 2008 17:17:12 +0000
Subject: changed queue object Construction/Startup interface

---
 msg.c     |  3 +--
 msg.h     |  1 -
 queue.c   | 31 ++++++++++++++++++++++---------
 queue.h   |  1 +
 syslogd.c |  6 +++++-
 5 files changed, 29 insertions(+), 13 deletions(-)

diff --git a/msg.c b/msg.c
index 8c02ed18..e2be7adc 100644
--- a/msg.c
+++ b/msg.c
@@ -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");
diff --git a/msg.h b/msg.h
index 467562fa..2168fb4c 100644
--- a/msg.h
+++ b/msg.h
@@ -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);
diff --git a/queue.c b/queue.c
index 8ef20371..d324172e 100644
--- a/queue.c
+++ b/queue.c
@@ -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)
 {
diff --git a/queue.h b/queue.h
index b62a4804..7ad9b1e1 100644
--- a/queue.h
+++ b/queue.h
@@ -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 */
diff --git a/syslogd.c b/syslogd.c
index 13ff8960..abb6d16b 100644
--- a/syslogd.c
+++ b/syslogd.c
@@ -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;
 
-- 
cgit v1.2.3