Merge branch 'c1-streamzip' into master-streamzip

Conflicts:
	ChangeLog
	configure.ac
	doc/rsyslog_conf_global.html
	runtime/stream.c
	tools/omfile.c

5 files changed, 125 insertions, 51 deletions

diff --git a/ChangeLog b/ChangeLog
index cdef134f..7a9b0e9b 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -6,6 +6,9 @@ Version 7.3.0  [devel] 2012-10-??
   * driverdirectory is now cleanly a global parameter, but can no longer
     be specified as an action paramter. Note that in previous versions
     this parameter was ignored in all but the first action definition
+- improved omfile zip writer to increase compression
+  This was achieved by somewhat reducing the robustness of the zip archive.
+  This is controlled by the new action parameter "VeryReliableZip".
 ---------------------------------------------------------------------------
 Version 7.1.9  [beta] 2012-10-??
 - bugfix: comments inside objects (e.g. action()) were not properly handled
diff --git a/doc/omfile.html b/doc/omfile.html
index 23ecc034..e58ea3b0 100644
--- a/doc/omfile.html
+++ b/doc/omfile.html
@@ -28,6 +28,16 @@
 	<li><strong>ZipLevel </strong>0..9 [default 0]<br>
 	if greater 0, turns on gzip compression of the output file. The higher the number, the better the compression, but also the more CPU is required for zipping.<br></li><br>
 
+	<li><b>VeryReliableZip</b> [<b>on</b>/off] (v7.3.0+) - if ZipLevel is greater 0, 
+	then this setting controls if extra headers are written to make the resulting file
+	extra hardened against malfunction. If set to off, data appended to previously unclean
+	closed files may not be accessible without extra tools.
+	Note that this risk is usually expected to be bearable, and thus "off" is the default mode.
+	The extra headers considerably
+	degrade compression, files with this option set to "on" may be four to five times as
+	large as files processed in "off" mode.
+	</li><br>
+
 	<li><strong>FlushInterval </strong>(not mandatory, default will be used)<br>
 	Defines a template to be used for the output. <br></li><br>
 
diff --git a/runtime/stream.c b/runtime/stream.c
index 742799d2..cc69ce5d 100644
--- a/runtime/stream.c
+++ b/runtime/stream.c
@@ -74,11 +74,12 @@ DEFobjStaticHelpers
 DEFobjCurrIf(zlibw)
 
 /* forward definitions */
-static rsRetVal strmFlushInternal(strm_t *pThis);
+static rsRetVal strmFlushInternal(strm_t *pThis, int bFlushZip);
 static rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf);
 static rsRetVal strmCloseFile(strm_t *pThis);
 static void *asyncWriterThread(void *pPtr);
-static rsRetVal doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf);
+static rsRetVal doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush);
+static rsRetVal doZipFinish(strm_t *pThis);
 static rsRetVal strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf);
 
 
@@ -341,7 +342,7 @@ static rsRetVal strmCloseFile(strm_t *pThis)
 		  (pThis->pszFName == NULL) ? "N/A" : (char*)pThis->pszFName);
 
 	if(pThis->tOperationsMode != STREAMMODE_READ) {
-		strmFlushInternal(pThis);
+		strmFlushInternal(pThis, 0);
 		if(pThis->bAsyncWrite) {
 			strmWaitAsyncWriterDone(pThis);
 		}
@@ -675,6 +676,7 @@ BEGINobjConstruct(strm) /* be sure to specify the object type also in END macro!
 	pThis->fd = -1;
 	pThis->fdDir = -1;
 	pThis->iUngetC = -1;
+	pThis->bVeryReliableZip = 0;
 	pThis->sType = STREAMTYPE_FILE_SINGLE;
 	pThis->sIOBufSize = glblGetIOBufSize();
 	pThis->tOpenMode = 0600;
@@ -777,6 +779,10 @@ stopWriter(strm_t *pThis)
 BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */
 	int i;
 CODESTARTobjDestruct(strm)
+	/* we need to stop the ZIP writer */
+	if(pThis->iZipLevel) {
+		doZipFinish(pThis);
+	}
 	if(pThis->bAsyncWrite)
 		/* Note: mutex will be unlocked in stopWriter! */
 		d_pthread_mutex_lock(&pThis->mut);
@@ -919,14 +925,14 @@ finalize_it:
 /* write memory buffer to a stream object.
  */
 static inline rsRetVal
-doWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+doWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush)
 {
 	DEFiRet;
 
 	ASSERT(pThis != NULL);
 
 	if(pThis->iZipLevel) {
-		CHKiRet(doZipWrite(pThis, pBuf, lenBuf));
+		CHKiRet(doZipWrite(pThis, pBuf, lenBuf, bFlush));
 	} else {
 		/* write without zipping */
 		CHKiRet(strmPhysWrite(pThis, pBuf, lenBuf));
@@ -971,7 +977,7 @@ doAsyncWriteInternal(strm_t *pThis, size_t lenBuf)
  * the background thread. -- rgerhards, 2009-07-07
  */
 static rsRetVal
-strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlushZip)
 {
 	DEFiRet;
 
@@ -990,7 +996,7 @@ strmSchedWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 	if(pThis->bAsyncWrite) {
 		CHKiRet(doAsyncWriteInternal(pThis, lenBuf));
 	} else {
-		CHKiRet(doWriteInternal(pThis, pBuf, lenBuf));
+		CHKiRet(doWriteInternal(pThis, pBuf, lenBuf, bFlushZip));
 	}
 
 
@@ -1030,7 +1036,7 @@ asyncWriterThread(void *pPtr)
 			}
 			if(bTimedOut && pThis->iBufPtr > 0) {
 				/* if we timed out, we need to flush pending data */
-				strmFlushInternal(pThis);
+				strmFlushInternal(pThis, 0);
 				bTimedOut = 0;
 				continue; /* now we should have data */
 			}
@@ -1056,7 +1062,7 @@ asyncWriterThread(void *pPtr)
 		bTimedOut = 0; /* we may have timed out, but there *is* work to do... */
 
 		iDeq = pThis->iDeq++ % STREAM_ASYNC_NUMBUFS;
-		doWriteInternal(pThis, pThis->asyncBuf[iDeq].pBuf, pThis->asyncBuf[iDeq].lenBuf);
+		doWriteInternal(pThis, pThis->asyncBuf[iDeq].pBuf, pThis->asyncBuf[iDeq].lenBuf, 0); // TODO: flush state
 		// TODO: error check????? 2009-07-06
 
 		--pThis->iCnt;
@@ -1166,63 +1172,102 @@ finalize_it:
  * rgerhards, 2009-06-04
  */
 static rsRetVal
-doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush)
 {
-	z_stream zstrm;
 	int zRet;	/* zlib return state */
 	sbool bzInitDone = RSFALSE;
 	DEFiRet;
+	unsigned outavail;
 	assert(pThis != NULL);
 	assert(pBuf != NULL);
 
-	/* allocate deflate state */
-	zstrm.zalloc = Z_NULL;
-	zstrm.zfree = Z_NULL;
-	zstrm.opaque = Z_NULL;
-	zstrm.next_in = (Bytef*) pBuf;	/* as of zlib doc, this must be set BEFORE DeflateInit2 */
-	/* see note in file header for the params we use with deflateInit2() */
-	zRet = zlibw.DeflateInit2(&zstrm, pThis->iZipLevel, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY);
-	if(zRet != Z_OK) {
-		DBGPRINTF("error %d returned from zlib/deflateInit2()\n", zRet);
-		ABORT_FINALIZE(RS_RET_ZLIB_ERR);
+	if(!pThis->bzInitDone) {
+		/* allocate deflate state */
+		pThis->zstrm.zalloc = Z_NULL;
+		pThis->zstrm.zfree = Z_NULL;
+		pThis->zstrm.opaque = Z_NULL;
+		/* see note in file header for the params we use with deflateInit2() */
+		zRet = zlibw.DeflateInit2(&pThis->zstrm, pThis->iZipLevel, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY);
+		if(zRet != Z_OK) {
+			DBGPRINTF("error %d returned from zlib/deflateInit2()\n", zRet);
+			ABORT_FINALIZE(RS_RET_ZLIB_ERR);
+		}
+		pThis->bzInitDone = RSTRUE;
 	}
 	bzInitDone = RSTRUE;
 
 	/* now doing the compression */
-	zstrm.next_in = (Bytef*) pBuf;	/* as of zlib doc, this must be set BEFORE DeflateInit2 */
-	zstrm.avail_in = lenBuf;
+	pThis->zstrm.next_in = (Bytef*) pBuf;
+	pThis->zstrm.avail_in = lenBuf;
 	/* run deflate() on buffer until everything has been compressed */
 	do {
-		DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", zstrm.avail_in, zstrm.total_in);
-		zstrm.avail_out = pThis->sIOBufSize;
-		zstrm.next_out = pThis->pZipBuf;
-		zRet = zlibw.Deflate(&zstrm, Z_FINISH);    /* no bad return value */
-		DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, zstrm.avail_out);
-		assert(zRet != Z_STREAM_ERROR);  /* state not clobbered */
-		if(zstrm.avail_out == pThis->sIOBufSize)
-			break; /* this is valid, indicates end of compression --> see zlib howto */
-		CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, pThis->sIOBufSize - zstrm.avail_out));
-	} while (zstrm.avail_out == 0);
-	assert(zstrm.avail_in == 0);     /* all input will be used */
+		DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", pThis->zstrm.avail_in, pThis->zstrm.total_in);
+		pThis->zstrm.avail_out = pThis->sIOBufSize;
+		pThis->zstrm.next_out = pThis->pZipBuf;
+		zRet = zlibw.Deflate(&pThis->zstrm, bFlush ? Z_SYNC_FLUSH : Z_NO_FLUSH);    /* no bad return value */
+		DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pThis->zstrm.avail_out);
+		outavail =pThis->sIOBufSize - pThis->zstrm.avail_out;
+		if(outavail != 0) {
+			CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, outavail));
+		}
+	} while (pThis->zstrm.avail_out == 0);
 
 finalize_it:
-	if(bzInitDone) {
-		zRet = zlibw.DeflateEnd(&zstrm);
-		if(zRet != Z_OK) {
-			DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet);
+	if(pThis->bzInitDone && pThis->bVeryReliableZip) {
+		doZipFinish(pThis);
+	}
+	RETiRet;
+}
+
+
+
+/* finish zlib buffer, to be called before closing the ZIP file (if
+ * running in stream mode).
+ */
+static rsRetVal
+doZipFinish(strm_t *pThis)
+{
+	int zRet;	/* zlib return state */
+	DEFiRet;
+	unsigned outavail;
+	assert(pThis != NULL);
+	assert(pBuf != NULL);
+
+	if(!pThis->bzInitDone) {
+		FINALIZE;
+	}
+
+dbgprintf("AAAA: doZipFinish() called\n");
+	pThis->zstrm.avail_in = 0;
+	/* run deflate() on buffer until everything has been compressed */
+	do {
+		DBGPRINTF("in deflate() loop, avail_in %d, total_in %ld\n", pThis->zstrm.avail_in, pThis->zstrm.total_in);
+		pThis->zstrm.avail_out = pThis->sIOBufSize;
+		pThis->zstrm.next_out = pThis->pZipBuf;
+		zRet = zlibw.Deflate(&pThis->zstrm, Z_FINISH);    /* no bad return value */
+		DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pThis->zstrm.avail_out);
+		outavail = pThis->sIOBufSize - pThis->zstrm.avail_out;
+		if(outavail != 0) {
+			CHKiRet(strmPhysWrite(pThis, (uchar*)pThis->pZipBuf, outavail));
 		}
+	} while (pThis->zstrm.avail_out == 0);
+
+finalize_it:
+	zRet = zlibw.DeflateEnd(&pThis->zstrm);
+	if(zRet != Z_OK) {
+		DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet);
 	}
 
+	pThis->bzInitDone = 0;
 	RETiRet;
 }
 
-
 /* flush stream output buffer to persistent storage. This can be called at any time
  * and is automatically called when the output buffer is full.
  * rgerhards, 2008-01-10
  */
 static rsRetVal
-strmFlushInternal(strm_t *pThis)
+strmFlushInternal(strm_t *pThis, int bFlushZip)
 {
 	DEFiRet;
 
@@ -1232,7 +1277,7 @@ strmFlushInternal(strm_t *pThis)
 		  (long) pThis->iBufPtr, (pThis->iBufPtr == 0) ? " (no need to flush)" : "");
 
 	if(pThis->tOperationsMode != STREAMMODE_READ && pThis->iBufPtr > 0) {
-		iRet = strmSchedWrite(pThis, pThis->pIOBuf, pThis->iBufPtr);
+		iRet = strmSchedWrite(pThis, pThis->pIOBuf, pThis->iBufPtr, bFlushZip);
 	}
 
 	RETiRet;
@@ -1254,7 +1299,7 @@ strmFlush(strm_t *pThis)
 
 	if(pThis->bAsyncWrite)
 		d_pthread_mutex_lock(&pThis->mut);
-	CHKiRet(strmFlushInternal(pThis));
+	CHKiRet(strmFlushInternal(pThis, 1));
 
 finalize_it:
 	if(pThis->bAsyncWrite)
@@ -1277,7 +1322,7 @@ static rsRetVal strmSeek(strm_t *pThis, off64_t offs)
 	if(pThis->fd == -1) {
 		CHKiRet(strmOpenFile(pThis));
 	} else {
-		CHKiRet(strmFlushInternal(pThis));
+		CHKiRet(strmFlushInternal(pThis, 0));
 	}
 	long long i;
 	DBGOPRINT((obj_t*) pThis, "file %d seek, pos %llu\n", pThis->fd, (long long unsigned) offs);
@@ -1320,7 +1365,7 @@ static rsRetVal strmWriteChar(strm_t *pThis, uchar c)
 
 	/* if the buffer is full, we need to flush before we can write */
 	if(pThis->iBufPtr == pThis->sIOBufSize) {
-		CHKiRet(strmFlushInternal(pThis));
+		CHKiRet(strmFlushInternal(pThis, 0));
 	}
 	/* we now always have space for one character, so we simply copy it */
 	*(pThis->pIOBuf + pThis->iBufPtr) = c;
@@ -1390,7 +1435,7 @@ strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 	iOffset = 0;
 	do {
 		if(pThis->iBufPtr == pThis->sIOBufSize) {
-			CHKiRet(strmFlushInternal(pThis)); /* get a new buffer for rest of data */
+			CHKiRet(strmFlushInternal(pThis, 0)); /* get a new buffer for rest of data */
 		}
 		iWrite = pThis->sIOBufSize - pThis->iBufPtr; /* this fits in current buf */
 		if(iWrite > lenBuf)
@@ -1405,7 +1450,7 @@ strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 	 * write it. This seems more natural than waiting (hours?) for the next message...
 	 */
 	if(pThis->iBufPtr == pThis->sIOBufSize) {
-		CHKiRet(strmFlushInternal(pThis)); /* get a new buffer for rest of data */
+		CHKiRet(strmFlushInternal(pThis, 0)); /* get a new buffer for rest of data */
 	}
 
 finalize_it:
@@ -1433,6 +1478,7 @@ DEFpropSetMeth(strm, tOperationsMode, int)
 DEFpropSetMeth(strm, tOpenMode, mode_t)
 DEFpropSetMeth(strm, sType, strmType_t)
 DEFpropSetMeth(strm, iZipLevel, int)
+DEFpropSetMeth(strm, bVeryReliableZip, int)
 DEFpropSetMeth(strm, bSync, int)
 DEFpropSetMeth(strm, sIOBufSize, size_t)
 DEFpropSetMeth(strm, iSizeLimit, off_t)
@@ -1564,7 +1610,7 @@ static rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm)
 	ISOBJ_TYPE_assert(pThis, strm);
 	ISOBJ_TYPE_assert(pStrm, strm);
 
-	strmFlushInternal(pThis);
+	strmFlushInternal(pThis, 0);
 	CHKiRet(obj.BeginSerialize(pStrm, (obj_t*) pThis));
 
 	objSerializeSCALAR(pStrm, iCurrFNum, INT);
@@ -1757,6 +1803,7 @@ CODESTARTobjQueryInterface(strm)
 	pIf->SettOpenMode = strmSettOpenMode;
 	pIf->SetsType = strmSetsType;
 	pIf->SetiZipLevel = strmSetiZipLevel;
+	pIf->SetbVeryReliableZip = strmSetbVeryReliableZip;
 	pIf->SetbSync = strmSetbSync;
 	pIf->SetsIOBufSize = strmSetsIOBufSize;
 	pIf->SetiSizeLimit = strmSetiSizeLimit;
diff --git a/runtime/stream.h b/runtime/stream.h
index a01929f2..fdfefaa3 100644
--- a/runtime/stream.h
+++ b/runtime/stream.h
@@ -124,6 +124,8 @@ typedef struct strm_s {
 	sbool bAsyncWrite;	/* do asynchronous writes (always if a flush interval is given) */
 	sbool bStopWriter;	/* shall writer thread terminate? */
 	sbool bDoTimedWait;	/* instruct writer thread to do a times wait to support flush timeouts */
+	sbool bzInitDone; /* did we do an init of zstrm already? */
+	sbool bVeryReliableZip; /* shall we write interim headers to create a very reliable ZIP file? */
 	int iFlushInterval; /* flush in which interval - 0, no flushing */
 	pthread_mutex_t mut;/* mutex for flush in async mode */
 	pthread_cond_t notFull;
@@ -132,6 +134,7 @@ typedef struct strm_s {
 	unsigned short iEnq;	/* this MUST be unsigned as we use module arithmetic (else invalid indexing happens!) */
 	unsigned short iDeq;	/* this MUST be unsigned as we use module arithmetic (else invalid indexing happens!) */
 	short iCnt;	/* current nbr of elements in buffer */
+	z_stream zstrm;	/* zip stream to use */
 	struct {
 		uchar *pBuf;
 		size_t lenBuf;
@@ -181,8 +184,10 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */
 	INTERFACEpropSetMeth(strm, pszSizeLimitCmd, uchar*);
 	/* v6 added */
 	rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr, int mode);
+	/* v7 added  2012-09-14 */
+	INTERFACEpropSetMeth(strm, bVeryReliableZip, int);
 ENDinterface(strm)
-#define strmCURR_IF_VERSION 6 /* increment whenever you change the interface structure! */
+#define strmCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */
 
 
 /* prototypes */
diff --git a/tools/omfile.c b/tools/omfile.c
index 1ccd4a56..2bb51e99 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -156,6 +156,7 @@ typedef struct _instanceData {
 	int	iFlushInterval;		/* how fast flush buffer on inactivity? */
 	sbool	bFlushOnTXEnd;		/* flush write buffers when transaction has ended? */
 	sbool	bUseAsyncWriter;	/* use async stream writer? */
+	sbool	bVeryRobustZip;
 } instanceData;
 
 
@@ -205,6 +206,7 @@ static struct cnfparamdescr actpdescr[] = {
 	{ "ziplevel", eCmdHdlrInt, 0 }, /* legacy: omfileziplevel */
 	{ "flushinterval", eCmdHdlrInt, 0 }, /* legacy: omfileflushinterval */
 	{ "asyncwriting", eCmdHdlrBinary, 0 }, /* legacy: omfileasyncwriting */
+	{ "veryrobustzip", eCmdHdlrBinary, 0 },
 	{ "flushontxend", eCmdHdlrBinary, 0 }, /* legacy: omfileflushontxend */
 	{ "iobuffersize", eCmdHdlrSize, 0 }, /* legacy: omfileiobuffersize */
 	{ "dirowner", eCmdHdlrUID, 0 }, /* legacy: dirowner */
@@ -269,7 +271,8 @@ CODESTARTdbgPrintInstInfo
 	dbgprintf("\tflush on TX end=%d\n", pData->bFlushOnTXEnd);
 	dbgprintf("\tflush interval=%d\n", pData->iFlushInterval);
 	dbgprintf("\tfile cache size=%d\n", pData->iDynaFileCacheSize);
-	dbgprintf("\tcreate directories: %s\n", pData->bCreateDirs ? "yes" : "no");
+	dbgprintf("\tcreate directories: %s\n", pData->bCreateDirs ? "on" : "off");
+	dbgprintf("\tvery robust zip: %s\n", pData->bCreateDirs ? "on" : "off");
 	dbgprintf("\tfile owner %d, group %d\n", (int) pData->fileUID, (int) pData->fileGID);
 	dbgprintf("\tdirectory owner %d, group %d\n", (int) pData->dirUID, (int) pData->dirGID);
 	dbgprintf("\tdir create mode 0%3.3o, file create mode 0%3.3o\n",
@@ -292,7 +295,7 @@ setLegacyDfltTpl(void __attribute__((unused)) *pVal, uchar* newVal)
 
 	if(loadModConf != NULL && loadModConf->tplName != NULL) {
 		free(newVal);
-		errmsg.LogError(0, RS_RET_ERR, "omfile default template already set via module "
+		errmsg.LogError(0, RS_RET_ERR, "omfile: default template already set via module "
 			"global parameter - can no longer be changed");
 		ABORT_FINALIZE(RS_RET_ERR);
 	}
@@ -536,6 +539,7 @@ prepareFile(instanceData *pData, uchar *newFileName)
 	CHKiRet(strm.SetFName(pData->pStrm, szBaseName, ustrlen(szBaseName)));
 	CHKiRet(strm.SetDir(pData->pStrm, szDirName, ustrlen(szDirName)));
 	CHKiRet(strm.SetiZipLevel(pData->pStrm, pData->iZipLevel));
+	CHKiRet(strm.SetbVeryReliableZip(pData->pStrm, pData->bVeryRobustZip));
 	CHKiRet(strm.SetsIOBufSize(pData->pStrm, (size_t) pData->iIOBufSize));
 	CHKiRet(strm.SettOperationsMode(pData->pStrm, STREAMMODE_WRITE_APPEND));
 	CHKiRet(strm.SettOpenMode(pData->pStrm, cs.fCreateMode));
@@ -842,6 +846,7 @@ BEGINendTransaction
 CODESTARTendTransaction
 	/* Note: pStrm may be NULL if there was an error opening the stream */
 	if(pData->bFlushOnTXEnd && pData->pStrm != NULL) {
+dbgprintf("AAAA: flusing stream, endTx\n");
 		CHKiRet(strm.Flush(pData->pStrm));
 	}
 finalize_it:
@@ -853,6 +858,7 @@ CODESTARTdoAction
 	DBGPRINTF("file to log to: %s\n", pData->f_fname);
 	CHKiRet(writeFile(ppString, iMsgOpts, pData));
 	if(!bCoreSupportsBatching && pData->bFlushOnTXEnd) {
+dbgprintf("AAAA: flusing stream, in Tx\n");
 		CHKiRet(strm.Flush(pData->pStrm));
 	}
 finalize_it:
@@ -877,6 +883,7 @@ setInstParamDefaults(instanceData *pData)
 	pData->bCreateDirs = 1;
 	pData->bSyncFile = 0;
 	pData->iZipLevel = 0;
+	pData->bVeryRobustZip = 0;
 	pData->bFlushOnTXEnd = FLUSHONTX_DFLT;
 	pData->iIOBufSize = IOBUF_DFLT_SIZE;
 	pData->iFlushInterval = FLUSH_INTRVL_DFLT;
@@ -914,6 +921,8 @@ CODESTARTnewActInst
 			pData->iZipLevel = (int) pvals[i].val.d.n;
 		} else if(!strcmp(actpblk.descr[i].name, "flushinterval")) {
 			pData->iFlushInterval = pvals[i].val.d.n;
+		} else if(!strcmp(actpblk.descr[i].name, "veryrobustzip")) {
+			pData->bVeryRobustZip = pvals[i].val.d.n;
 		} else if(!strcmp(actpblk.descr[i].name, "asyncwriting")) {
 			pData->bUseAsyncWriter = pvals[i].val.d.n;
 		} else if(!strcmp(actpblk.descr[i].name, "flushontxend")) {
@@ -1063,6 +1072,7 @@ CODESTARTparseSelectorAct
 	pData->iIOBufSize = (int) cs.iIOBufSize;
 	pData->iFlushInterval = cs.iFlushInterval;
 	pData->bUseAsyncWriter = cs.bUseAsyncWriter;
+	pData->bVeryRobustZip = 0;	/* cannot be specified via legacy conf */
 CODE_STD_FINALIZERparseSelectorAct
 ENDparseSelectorAct
 
@@ -1089,7 +1099,6 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
 	cs.bUseAsyncWriter = USE_ASYNCWRITER_DFLT;
 	free(pszFileDfltTplName);
 	pszFileDfltTplName = NULL;
-
 	return RS_RET_OK;
 }