Merge branch 'v4-beta' into tmp

Conflicts:
	ChangeLog

4 files changed, 28 insertions, 14 deletions

diff --git a/runtime/conf.c b/runtime/conf.c
index 83ed2e9b..b92664a1 100644
--- a/runtime/conf.c
+++ b/runtime/conf.c
@@ -568,6 +568,7 @@ finalize_it:
  * rgerhards, 2007-07-25
  * updated to include OMSR pointer -- rgerhards, 2007-07-27
  * updated to include template name -- rgerhards, 2008-03-28
+ * rgerhards, 2010-01-19: file names end at the first space
  */
 rsRetVal
 cflineParseFileName(uchar* p, uchar *pFileName, omodStringRequest_t *pOMSR, int iEntry, int iTplOpts, uchar *pszTpl)
@@ -580,7 +581,7 @@ cflineParseFileName(uchar* p, uchar *pFileName, omodStringRequest_t *pOMSR, int
 
 	pName = pFileName;
 	i = 1; /* we start at 1 so that we reseve space for the '\0'! */
-	while(*p && *p != ';' && i < MAXFNAME) {
+	while(*p && *p != ';' && *p != ' ' && i < MAXFNAME) {
 		*pName++ = *p++;
 		++i;
 	}
diff --git a/runtime/queue.c b/runtime/queue.c
index ddff1bcf..9d7a9058 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -858,9 +858,11 @@ static rsRetVal qDestructDisk(qqueue_t *pThis)
 	DEFiRet;
 	
 	ASSERT(pThis != NULL);
-	
-	strm.Destruct(&pThis->tVars.disk.pWrite);
-	strm.Destruct(&pThis->tVars.disk.pRead);
+
+	if (pThis->tVars.disk.pWrite != NULL)
+		strm.Destruct(&pThis->tVars.disk.pWrite);
+	if (pThis->tVars.disk.pRead != NULL)
+		strm.Destruct(&pThis->tVars.disk.pRead);
 
 	RETiRet;
 }
@@ -1214,7 +1216,7 @@ static rsRetVal qqueueShutdownWorkers(qqueue_t *pThis)
 			/* we need to re-aquire the mutex for the next check in this case! */
 			BEGIN_MTX_PROTECTED_OPERATIONS(pThis->mut, LOCK_MUTEX);	/* some workers may be running in parallel! */
 		}
-		if(pThis->bIsDA && wtpGetCurNumWrkr(pThis->pWtpDA, LOCK_MUTEX) > 0) {
+		if(pThis->bRunsDA && wtpGetCurNumWrkr(pThis->pWtpDA, LOCK_MUTEX) > 0) {
 			/* and now the same for the DA queue */
 			END_MTX_PROTECTED_OPERATIONS(pThis->mut);
 			dbgoprint((obj_t*) pThis, "trying immediate shutdown of DA workers\n");
@@ -1666,7 +1668,7 @@ qqueueChkStopWrkrDA(qqueue_t *pThis)
 	if(pThis->bEnqOnly) {
 		bStopWrkr = 1;
 	} else {
-		if(pThis->bRunsDA) {
+		if(pThis->bRunsDA == 2) {
 			ASSERT(pThis->pqDA != NULL);
 			if(   pThis->pqDA->bEnqOnly
 			   && pThis->pqDA->sizeOnDiskMax > 0
@@ -1921,7 +1923,8 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint)
 			pThis->bNeedDelQIF = 0;
 		}
 		/* indicate spool file needs to be deleted */
-		CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pRead, 1));
+		if (pThis->tVars.disk.pRead != NULL)
+			CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pRead, 1));
 		FINALIZE; /* nothing left to do, so be happy */
 	}
 
@@ -1955,13 +1958,15 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint)
 	}
 
 	/* now persist the stream info */
-	CHKiRet(strm.Serialize(pThis->tVars.disk.pWrite, psQIF));
-	CHKiRet(strm.Serialize(pThis->tVars.disk.pRead, psQIF));
+	if (pThis->tVars.disk.pWrite != NULL)
+		CHKiRet(strm.Serialize(pThis->tVars.disk.pWrite, psQIF));
+	if (pThis->tVars.disk.pRead != NULL)
+		CHKiRet(strm.Serialize(pThis->tVars.disk.pRead, psQIF));
 	
 	/* tell the input file object that it must not delete the file on close if the queue
 	 * is non-empty - but only if we are not during a simple checkpoint
 	 */
-	if(bIsCheckpoint != QUEUE_CHECKPOINT) {
+	if(bIsCheckpoint != QUEUE_CHECKPOINT && pThis->tVars.disk.pRead != NULL) {
 		CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pRead, 0));
 	}
 
diff --git a/runtime/stream.c b/runtime/stream.c
index 2d1e9380..36f44003 100644
--- a/runtime/stream.c
+++ b/runtime/stream.c
@@ -209,13 +209,19 @@ doPhysOpen(strm_t *pThis)
 		default:assert(0);
 			break;
 	}
+	if(pThis->sType == STREAMTYPE_NAMED_PIPE) {
+		DBGPRINTF("Note: stream '%s' is a named pipe, open with O_NONBLOCK\n", pThis->pszCurrFName);
+		iFlags |= O_NONBLOCK;
+	}
 
 	pThis->fd = open((char*)pThis->pszCurrFName, iFlags, pThis->tOpenMode);
 	DBGPRINTF("file '%s' opened as #%d with mode %d\n", pThis->pszCurrFName, pThis->fd, pThis->tOpenMode);
 	if(pThis->fd == -1) {
-		int ierrnoSave = errno;
-		dbgoprint((obj_t*) pThis, "open error %d, file '%s'\n", errno, pThis->pszCurrFName);
-		if(ierrnoSave == ENOENT)
+		char errStr[1024];
+		int err = errno;
+		rs_strerror_r(err, errStr, sizeof(errStr));
+		dbgoprint((obj_t*) pThis, "open error %d, file '%s': %s\n", errno, pThis->pszCurrFName, errStr);
+		if(err == ENOENT)
 			ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND);
 		else
 			ABORT_FINALIZE(RS_RET_IO_ERROR);
@@ -429,6 +435,7 @@ strmHandleEOF(strm_t *pThis)
 	ISOBJ_TYPE_assert(pThis, strm);
 	switch(pThis->sType) {
 		case STREAMTYPE_FILE_SINGLE:
+		case STREAMTYPE_NAMED_PIPE:
 			ABORT_FINALIZE(RS_RET_EOF);
 			break;
 		case STREAMTYPE_FILE_CIRCULAR:
diff --git a/runtime/stream.h b/runtime/stream.h
index 64ffb6e1..89175b0f 100644
--- a/runtime/stream.h
+++ b/runtime/stream.h
@@ -76,7 +76,8 @@
 typedef enum {
 	STREAMTYPE_FILE_SINGLE = 0,	/**< read a single file */
 	STREAMTYPE_FILE_CIRCULAR = 1,	/**< circular files */
-	STREAMTYPE_FILE_MONITOR = 2	/**< monitor a (third-party) file */
+	STREAMTYPE_FILE_MONITOR = 2,	/**< monitor a (third-party) file */
+	STREAMTYPE_NAMED_PIPE = 3	/**< file is a named pipe (so far, tested for output only) */
 } strmType_t;
 
 typedef enum {				/* when extending, do NOT change existing modes! */