modified stream class and omfile to work with it

now some basic operations are carried out via the stream class.

5 files changed, 225 insertions, 54 deletions

diff --git a/doc/rsyslog_conf_global.html b/doc/rsyslog_conf_global.html
index 778e18f8..0f408fcf 100644
--- a/doc/rsyslog_conf_global.html
+++ b/doc/rsyslog_conf_global.html
@@ -189,6 +189,9 @@ supported in order to be compliant to the upcoming new syslog RFC series.
 <li><a href="rsconf1_maxopenfiles.html">$MaxOpenFiles</a></li>
 <li><a href="rsconf1_moddir.html">$ModDir</a></li>
 <li><a href="rsconf1_modload.html">$ModLoad</a></li>
+<li><b>$OMFileZipLevel</b> 0..9  [default 0] - 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.</li>
 <li><b>$RepeatedMsgContainsOriginalMsg</b> [on/<b>off</b>] - "last message repeated n times" messages, if generated,
 have a different format that contains the message that is being repeated.
 Note that only the first "n" characters are included, with n to be at least 80 characters, most
diff --git a/runtime/queue.c b/runtime/queue.c
index 3f14b535..3532a145 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -1923,8 +1923,7 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint)
 	}
 
 	CHKiRet(strm.Construct(&psQIF));
-	CHKiRet(strm.SettOperationsMode(psQIF, STREAMMODE_WRITE));
-	CHKiRet(strm.SetiAddtlOpenFlags(psQIF, O_TRUNC));
+	CHKiRet(strm.SettOperationsMode(psQIF, STREAMMODE_WRITE_TRUNC));
 	CHKiRet(strm.SetsType(psQIF, STREAMTYPE_FILE_SINGLE));
 	CHKiRet(strm.SetFName(psQIF, pszQIFNam, lenQIFNam));
 	CHKiRet(strm.ConstructFinalize(psQIF));
diff --git a/runtime/stream.c b/runtime/stream.c
index 261eb9ca..abcfce0c 100644
--- a/runtime/stream.c
+++ b/runtime/stream.c
@@ -1,4 +1,3 @@
-//TODO: O_TRUC mode!
 /* The serial stream class.
  *
  * A serial stream provides serial data access. In theory, serial streams
@@ -50,6 +49,7 @@
 
 /* static data */
 DEFobjStaticHelpers
+DEFobjCurrIf(zlibw)
 
 /* forward definitions */
 static rsRetVal strmFlush(strm_t *pThis);
@@ -74,7 +74,6 @@ static rsRetVal strmOpenFile(strm_t *pThis)
 	int iFlags;
 
 	ASSERT(pThis != NULL);
-	ASSERT(pThis->tOperationsMode == STREAMMODE_READ || pThis->tOperationsMode == STREAMMODE_WRITE);
 
 	if(pThis->fd != -1)
 		ABORT_FINALIZE(RS_RET_OK);
@@ -95,13 +94,27 @@ static rsRetVal strmOpenFile(strm_t *pThis)
 		}
 	}
 
+RUNLOG_VAR("%d", pThis->tOperationsMode);
 	/* compute which flags we need to provide to open */
-	if(pThis->tOperationsMode == STREAMMODE_READ)
-		iFlags = O_RDONLY;
-	else
-		iFlags = O_WRONLY | O_CREAT;
+	switch(pThis->tOperationsMode) {
+		case STREAMMODE_READ:
+			iFlags = O_CLOEXEC | O_NOCTTY | O_RDONLY;
+			break;
+		case STREAMMODE_WRITE:	/* legacy mode used inside queue engine */
+			iFlags = O_CLOEXEC | O_NOCTTY | O_WRONLY | O_CREAT;
+			break;
+		case STREAMMODE_WRITE_TRUNC:
+			iFlags = O_CLOEXEC | O_NOCTTY | O_WRONLY | O_CREAT | O_TRUNC;
+			break;
+		case STREAMMODE_WRITE_APPEND:
+			iFlags = O_CLOEXEC | O_NOCTTY | O_WRONLY | O_CREAT | O_APPEND;
+			break;
+		default:assert(0);
+			break;
+	}
 
-	iFlags |= pThis->iAddtlOpenFlags;
+	iFlags |= pThis->iAddtlOpenFlags; // TODO: remove this!
+dbgprintf("XXX: open with flags %d\n", iFlags);
 
 	pThis->fd = open((char*)pThis->pszCurrFName, iFlags, pThis->tOpenMode);
 	if(pThis->fd == -1) {
@@ -135,7 +148,7 @@ static rsRetVal strmCloseFile(strm_t *pThis)
 	ASSERT(pThis->fd != -1);
 	dbgoprint((obj_t*) pThis, "file %d closing\n", pThis->fd);
 
-	if(pThis->tOperationsMode == STREAMMODE_WRITE)
+	if(pThis->tOperationsMode != STREAMMODE_READ)
 		strmFlush(pThis);
 
 	close(pThis->fd); // TODO: error check
@@ -398,14 +411,25 @@ ENDobjConstruct(strm)
  */
 static rsRetVal strmConstructFinalize(strm_t *pThis)
 {
+	rsRetVal localRet;
 	DEFiRet;
 
 	ASSERT(pThis != NULL);
 
-	if(pThis->pIOBuf == NULL) { /* allocate our io buffer in case we have not yet */
-		if((pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)) == NULL)
-			ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
-		pThis->iBufPtrMax = 0; /* results in immediate read request */
+	CHKmalloc(pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize));
+	pThis->iBufPtrMax = 0; /* results in immediate read request */
+	if(pThis->iZipLevel) { /* do we need a zip buf? */
+		localRet = objUse(zlibw, LM_ZLIBW_FILENAME);
+		if(localRet != RS_RET_OK) {
+			pThis->iZipLevel = 0;
+			DBGPRINTF("stream was requested with zip mode, but zlibw module unavailable (%d) - using "
+				  "without zip\n", localRet);
+		} else {
+			/* we use the same size as the original buf, as we would like
+			 * to make sure we can write out everyting with a SINGLE api call!
+			 */
+			CHKmalloc(pThis->pZipBuf = (Bytef*) malloc(sizeof(uchar) * pThis->sIOBufSize));
+		}
 	}
 
 finalize_it:
@@ -416,15 +440,20 @@ finalize_it:
 /* destructor for the strm object */
 BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */
 CODESTARTobjDestruct(strm)
-	if(pThis->tOperationsMode == STREAMMODE_WRITE)
+	if(pThis->tOperationsMode != STREAMMODE_READ)
 		strmFlush(pThis);
 
 	/* ... then free resources */
 	if(pThis->fd != -1)
 		strmCloseFile(pThis);
 
+	if(pThis->iZipLevel) { /* do we need a zip buf? */
+		objRelease(zlibw, LM_ZLIBW_FILENAME);
+	}
+
 	free(pThis->pszDir);
 	free(pThis->pIOBuf);
+	free(pThis->pZipBuf);
 	free(pThis->pszCurrFName);
 	free(pThis->pszFName);
 ENDobjDestruct(strm)
@@ -453,20 +482,17 @@ finalize_it:
 }
 
 
-/* write memory buffer to a stream object.
- * To support direct writes of large objects, this method may be called
- * with a buffer pointing to some region other than the stream buffer itself.
- * However, in that case the stream buffer must be empty (strmFlush() has to
- * be called before), because we would otherwise mess up with the sequence
- * inside the stream. -- rgerhards, 2008-01-10
+/* physically write to the output file. the provided data is ready for
+ * writing (e.g. zipped if we are requested to do that).
+ * rgerhards, 2009-06-04
  */
-static rsRetVal strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+static rsRetVal
+strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 {
 	DEFiRet;
 	int iWritten;
 
 	ASSERT(pThis != NULL);
-	ASSERT(pBuf == pThis->pIOBuf || pThis->iBufPtr == 0);
 
 	if(pThis->fd == -1)
 		CHKiRet(strmOpenFile(pThis));
@@ -499,6 +525,95 @@ finalize_it:
 }
 
 
+/* write the output buffer in zip mode
+ * This means we compress it first and then do a physical write.
+ * Note that we always do a full deflateInit ... deflate ... deflateEnd
+ * sequence. While this is not optimal, we need to do it because we need
+ * to ensure that the file is readable even when we are aborted. Doing the
+ * full sequence brings us as far towards this goal as possible (and not
+ * doing it would be a total failure). It may be worth considering to
+ * add a config switch so that the user can decide the risk he is ready
+ * to take, but so far this is not yet implemented (not even requested ;)).
+ * rgerhards, 2009-06-04
+ */
+static rsRetVal
+doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+{
+	z_stream zstrm;
+	int zRet;	/* zlib return state */
+	DEFiRet;
+	assert(pThis != NULL);
+	assert(pBuf != NULL);
+
+	/* allocate deflate state */
+	zstrm.zalloc = Z_NULL;
+	zstrm.zfree = Z_NULL;
+	zstrm.opaque = Z_NULL;
+	/* 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);
+	}
+RUNLOG_STR("deflateInit2() done successfully\n");
+
+	/* now doing the compression */
+	zstrm.avail_in = lenBuf;
+	zstrm.next_in = (Bytef*) pBuf;
+	/* run deflate() on input until output buffer not full, finish
+	   compression if all of source has been read in */
+	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 */
+		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 */
+
+RUNLOG_STR("deflate() should be done successfully\n");
+
+	zRet = zlibw.DeflateEnd(&zstrm);
+	if(zRet != Z_OK) {
+		dbgprintf("error %d returned from zlib/deflateEnd()\n", zRet);
+		ABORT_FINALIZE(RS_RET_ZLIB_ERR);
+	}
+RUNLOG_STR("deflateEnd() done successfully\n");
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* write memory buffer to a stream object.
+ * To support direct writes of large objects, this method may be called
+ * with a buffer pointing to some region other than the stream buffer itself.
+ * However, in that case the stream buffer must be empty (strmFlush() has to
+ * be called before), because we would otherwise mess up with the sequence
+ * inside the stream. -- rgerhards, 2008-01-10
+ */
+static rsRetVal
+strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+{
+	DEFiRet;
+
+	ASSERT(pThis != NULL);
+	ASSERT(pBuf == pThis->pIOBuf || pThis->iBufPtr == 0);
+
+	if(pThis->iZipLevel) {
+		CHKiRet(doZipWrite(pThis, pBuf, lenBuf));
+	} else {
+		/* write without zipping */
+		CHKiRet(strmPhysWrite(pThis, pBuf, lenBuf));
+	}
+
+finalize_it:
+	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
@@ -511,7 +626,7 @@ strmFlush(strm_t *pThis)
 	ASSERT(pThis != NULL);
 	dbgoprint((obj_t*) pThis, "file %d flush, buflen %ld\n", pThis->fd, (long) pThis->iBufPtr);
 
-	if(pThis->tOperationsMode == STREAMMODE_WRITE && pThis->iBufPtr > 0) {
+	if(pThis->tOperationsMode != STREAMMODE_READ && pThis->iBufPtr > 0) {
 		iRet = strmWriteInternal(pThis, pThis->pIOBuf, pThis->iBufPtr);
 	}
 
@@ -605,6 +720,7 @@ strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
 	ASSERT(pThis != NULL);
 	ASSERT(pBuf != NULL);
 
+dbgprintf("strmWrite(%p, '%s', %ld);\n", pThis, pBuf,lenBuf);
 	/* check if the to-be-written data is larger than our buffer size */
 	if(lenBuf >= pThis->sIOBufSize) {
 		/* it is - so we do a direct write, that is most efficient.
@@ -646,6 +762,7 @@ DEFpropSetMeth(strm, iFileNumDigits, int)
 DEFpropSetMeth(strm, tOperationsMode, int)
 DEFpropSetMeth(strm, tOpenMode, mode_t)
 DEFpropSetMeth(strm, sType, strmType_t)
+DEFpropSetMeth(strm, iZipLevel, int)
 
 static rsRetVal strmSetiMaxFiles(strm_t *pThis, int iNewVal)
 {
@@ -681,13 +798,14 @@ strmSetFName(strm_t *pThis, uchar *pszName, size_t iLenName)
 	ASSERT(pThis != NULL);
 	ASSERT(pszName != NULL);
 	
+dbgprintf("XXX: strm setFname: '%s'\n", pszName);
 	if(iLenName < 1)
 		ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING);
 
 	if(pThis->pszFName != NULL)
 		free(pThis->pszFName);
 
-	if((pThis->pszFName = malloc(sizeof(uchar) * iLenName + 1)) == NULL)
+	if((pThis->pszFName = malloc(sizeof(uchar) * (iLenName + 1))) == NULL)
 		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
 
 	memcpy(pThis->pszFName, pszName, iLenName + 1); /* always think about the \0! */
@@ -711,6 +829,7 @@ strmSetDir(strm_t *pThis, uchar *pszDir, size_t iLenDir)
 	ASSERT(pThis != NULL);
 	ASSERT(pszDir != NULL);
 	
+dbgprintf("XXX: strm setDir: '%s'\n", pszDir);
 	if(iLenDir < 1)
 		ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING);
 
@@ -927,7 +1046,7 @@ CODESTARTobjQueryInterface(strm)
 	pIf->RecordBegin = strmRecordBegin;
 	pIf->RecordEnd = strmRecordEnd;
 	pIf->Serialize = strmSerialize;
-	pIf->SetiAddtlOpenFlags = strmSetiAddtlOpenFlags;
+	/* TODO: remove this! */ pIf->SetiAddtlOpenFlags = strmSetiAddtlOpenFlags;
 	pIf->GetCurrOffset = strmGetCurrOffset;
 	pIf->SetWCntr = strmSetWCntr;
 	/* set methods */
@@ -938,6 +1057,7 @@ CODESTARTobjQueryInterface(strm)
 	pIf->SettOperationsMode = strmSettOperationsMode;
 	pIf->SettOpenMode = strmSettOpenMode;
 	pIf->SetsType = strmSetsType;
+	pIf->SetiZipLevel = strmSetiZipLevel;
 finalize_it:
 ENDobjQueryInterface(strm)
 
diff --git a/runtime/stream.h b/runtime/stream.h
index ece33270..449bf6c6 100644
--- a/runtime/stream.h
+++ b/runtime/stream.h
@@ -47,6 +47,7 @@
 #include "obj-types.h"
 #include "glbl.h"
 #include "stream.h"
+#include "zlibw.h"
 
 /* stream types */
 typedef enum {
@@ -55,10 +56,12 @@ typedef enum {
 	STREAMTYPE_FILE_MONITOR = 2	/**< monitor a (third-party) file */
 } strmType_t;
 
-typedef enum {
+typedef enum {				/* when extending, do NOT change existing modes! */
 	STREAMMMODE_INVALID = 0,
 	STREAMMODE_READ = 1,
-	STREAMMODE_WRITE = 2
+	STREAMMODE_WRITE = 2,
+	STREAMMODE_WRITE_TRUNC = 3,
+	STREAMMODE_WRITE_APPEND = 4
 } strmMode_t;
 
 /* The strm_t data structure */
@@ -89,6 +92,9 @@ typedef struct strm_s {
 	size_t iBufPtr;	/* pointer into current buffer */
 	int iUngetC;	/* char set via UngetChar() call or -1 if none set */
 	int bInRecord;	/* if 1, indicates that we are currently writing a not-yet complete record */
+	int iZipLevel;	/* zip level (0..9). If 0, zip is completely disabled */
+	Bytef *pZipBuf;
+
 } strm_t;
 
 /* interfaces */
@@ -121,6 +127,7 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */
 	INTERFACEpropSetMeth(strm, tOperationsMode, int);
 	INTERFACEpropSetMeth(strm, tOpenMode, mode_t);
 	INTERFACEpropSetMeth(strm, sType, strmType_t);
+	INTERFACEpropSetMeth(strm, iZipLevel, int);
 ENDinterface(strm)
 #define strmCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
 
diff --git a/tools/omfile.c b/tools/omfile.c
index 689bad0c..3f517906 100644
--- a/tools/omfile.c
+++ b/tools/omfile.c
@@ -70,6 +70,7 @@
 #include <assert.h>
 #include <errno.h>
 #include <ctype.h>
+#include <libgen.h>
 #include <unistd.h>
 #include <sys/file.h>
 
@@ -87,6 +88,7 @@
 #include "module-template.h"
 #include "errmsg.h"
 #include "unicode-helper.h"
+#include "stream.h"
 #include "zlibw.h"
 
 MODULE_TYPE_OUTPUT
@@ -96,6 +98,7 @@ MODULE_TYPE_OUTPUT
 DEF_OMOD_STATIC_DATA
 DEFobjCurrIf(errmsg)
 DEFobjCurrIf(zlibw)
+DEFobjCurrIf(strm)
 
 /* The following structure is a dynafile name cache entry.
  */
@@ -131,12 +134,14 @@ static uid_t	dirUID;		/* UID to be used for newly created directories */
 static uid_t	dirGID;		/* GID to be used for newly created directories */
 static int	bCreateDirs;	/* auto-create directories for dynaFiles: 0 - no, 1 - yes */
 static int	bEnableSync = 0;/* enable syncing of files (no dash in front of pathname in conf): 0 - no, 1 - yes */
+static int	iZipLevel = 0;	/* zip compression mode (0..9 as usual) */
 static uchar	*pszTplName = NULL; /* name of the default template to use */
 /* end globals for default values */
 
 
 typedef struct _instanceData {
 	uchar	f_fname[MAXFNAME];/* file or template name (display only) */
+	strm_t	*pStrm;		/* our output stream */
 	short	fd;		/* file descriptor for (current) file */
 	outbuf_t *poBuf;	/* output buffer */
 	enum {
@@ -165,6 +170,7 @@ typedef struct _instanceData {
 	dynaFileCacheEntry **dynCache;
 	off_t	f_sizeLimit;		/* file size limit, 0 = no limit */
 	uchar	*f_sizeLimitCmd;	/* command to carry out when size limit is reached */
+	int 	iZipLevel;		/* zip mode to use for this selector */
 } instanceData;
 
 
@@ -446,13 +452,11 @@ prepareFile(instanceData *pData, uchar *newFileName)
 		FINALIZE; /* we are done in this case */
 	}
 
-	if(access((char*)newFileName, F_OK) == 0) {
-		/* file already exists */
-		pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY|O_CLOEXEC,
-				pData->fCreateMode);
-	} else {
-		pData->fd = -1;
+	// TODO: handle TTY case! (here or in stream.c?) 2009-06-04
+
+	if(access((char*)newFileName, F_OK) != 0) {
 		/* file does not exist, create it (and eventually parent directories */
+		pData->fd = -1;
 		if(pData->bCreateDirs) {
 			/* We first need to create parent dirs if they are missing.
 			 * We do not report any errors here ourselfs but let the code
@@ -485,8 +489,29 @@ prepareFile(instanceData *pData, uchar *newFileName)
 					 */
 				}
 			}
+			close(pData->fd); /* close again, as we need a stream further on */
 		}
 	}
+
+	char szNameBuf[MAXFNAME];
+	char szDirName[MAXFNAME];
+	char szBaseName[MAXFNAME];
+	strcpy(szNameBuf, (char*)pData->f_fname);
+	strcpy(szDirName, dirname(szNameBuf));
+	strcpy(szNameBuf, (char*)pData->f_fname);
+	strcpy(szBaseName, basename(szNameBuf));
+DBGPRINTF("XXX: name to set: '%s', dirname '%s'\n", pData->f_fname, szDirName);
+
+	CHKiRet(strm.Construct(&pData->pStrm));
+	CHKiRet(strm.SetFName(pData->pStrm, (uchar*)szBaseName, strlen(szBaseName)));
+	CHKiRet(strm.SetDir(pData->pStrm, (uchar*)szDirName, strlen(szDirName)));
+	CHKiRet(strm.SetiZipLevel(pData->pStrm, pData->iZipLevel));
+	CHKiRet(strm.SettOperationsMode(pData->pStrm, STREAMMODE_WRITE_APPEND));
+	CHKiRet(strm.SetsType(pData->pStrm, STREAMTYPE_FILE_SINGLE));
+	CHKiRet(strm.ConstructFinalize(pData->pStrm));
+	
+	pData->fd = 2; /* TODO: dummy to keep current inconsistent code happy - remove later */
+
 finalize_it:
 	/* this was "pData->fd != 0", which I think was a bug. I guess 0 was intended to mean
 	 * non-open file descriptor. Anyhow, I leave this comment for the time being to that if
@@ -715,20 +740,20 @@ static rsRetVal
 doZipWrite(instanceData *pData)
 {
 	outbuf_t *poBuf;
-	z_stream strm;
+	z_stream zstrm;
 	int zRet;	/* zlib return state */
 	DEFiRet;
 	assert(pData != NULL);
 
 	poBuf = pData->poBuf; 	/* use as a shortcut */
-	strm = poBuf->zStrm;	/* another shortcut */
+	zstrm = poBuf->zStrm;	/* another shortcut */
 
 	/* allocate deflate state */
-	strm.zalloc = Z_NULL;
-	strm.zfree = Z_NULL;
-	strm.opaque = Z_NULL;
+	zstrm.zalloc = Z_NULL;
+	zstrm.zfree = Z_NULL;
+	zstrm.opaque = Z_NULL;
 	/* see note in file header for the params we use with deflateInit2() */
-	zRet = zlibw.DeflateInit2(&strm, 9, Z_DEFLATED, 31, 9, Z_DEFAULT_STRATEGY);
+	zRet = zlibw.DeflateInit2(&zstrm, 9, 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);
@@ -736,24 +761,24 @@ doZipWrite(instanceData *pData)
 RUNLOG_STR("deflateInit2() done successfully\n");
 
 	/* now doing the compression */
-	strm.avail_in = poBuf->iBuf;
-	strm.next_in = (Bytef*) poBuf->pszBuf;
+	zstrm.avail_in = poBuf->iBuf;
+	zstrm.next_in = (Bytef*) poBuf->pszBuf;
 	/* run deflate() on input until output buffer not full, finish
 	   compression if all of source has been read in */
 	do {
-		dbgprintf("in deflate() loop, avail_in %d, total_in %ld\n", strm.avail_in, strm.total_in);
-		strm.avail_out = OUTBUF_LEN;
-		strm.next_out = (Bytef*) poBuf->zipBuf;
-		zRet = zlibw.Deflate(&strm, Z_FINISH);    /* no bad return value */
-		dbgprintf("after deflate, ret %d, avail_out %d\n", zRet, strm.avail_out);
+		dbgprintf("in deflate() loop, avail_in %d, total_in %ld\n", zstrm.avail_in, zstrm.total_in);
+		zstrm.avail_out = OUTBUF_LEN;
+		zstrm.next_out = (Bytef*) poBuf->zipBuf;
+		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 */
-		CHKiRet(doPhysWrite(pData, poBuf->fd, poBuf->zipBuf, OUTBUF_LEN - strm.avail_out));
-	} while (strm.avail_out == 0);
-	assert(strm.avail_in == 0);     /* all input will be used */
+		CHKiRet(doPhysWrite(pData, poBuf->fd, poBuf->zipBuf, OUTBUF_LEN - zstrm.avail_out));
+	} while (zstrm.avail_out == 0);
+	assert(zstrm.avail_in == 0);     /* all input will be used */
 
 RUNLOG_STR("deflate() should be done successfully\n");
 
-	zRet = zlibw.DeflateEnd(&strm);
+	zRet = zlibw.DeflateEnd(&zstrm);
 	if(zRet != Z_OK) {
 		dbgprintf("error %d returned from zlib/deflateEnd()\n", zRet);
 		ABORT_FINALIZE(RS_RET_ZLIB_ERR);
@@ -802,8 +827,14 @@ doWrite(instanceData *pData, uchar *pszBuf, int lenBuf)
 	ASSERT(pszBuf != NULL);
 
 	poBuf = pData->poBuf; 	/* use as a shortcut */
-dbgprintf("doWrite, pData->fd %d, poBuf->fd %d, iBuf %ld, lenBuf %ld\n",
-pData->fd, pData->poBuf->fd, pData->poBuf->iBuf, poBuf->lenBuf);
+dbgprintf("doWrite, pData->fd %d, pData->pStrm %p, poBuf->fd %d, iBuf %ld, lenBuf %ld\n",
+pData->fd, pData->pStrm, pData->poBuf->fd, pData->poBuf->iBuf, poBuf->lenBuf);
+
+RUNLOG_VAR("%p", pData->pStrm);
+	if(pData->pStrm != NULL){
+		CHKiRet(strm.Write(pData->pStrm, pszBuf, lenBuf));
+		FINALIZE; // TODO: clean up later
+	}
 
 	if(pData->fd != poBuf->fd) {
 		// TODO: more efficient use for dynafiles
@@ -869,6 +900,10 @@ ENDcreateInstance
 BEGINfreeInstance
 CODESTARTfreeInstance
 	doFlush(pData); /* flush anything that is pending, TODO: change when enhancing dynafile handling! */
+	if(pData->pStrm != NULL) {
+RUNLOG_STR("XXX: destructing stream");
+		strm.Destruct(&pData->pStrm);
+	}
 	if(pData->bDynamicName) {
 		dynaFileFreeCache(pData);
 	} else if(pData->fd != -1)
@@ -945,6 +980,7 @@ CODESTARTparseSelectorAct
 		pData->fileGID = fileGID;
 		pData->dirUID = dirUID;
 		pData->dirGID = dirGID;
+		pData->iZipLevel = iZipLevel;
 		pData->iDynaFileCacheSize = iDynaFileCacheSize; /* freeze current setting */
 		/* we now allocate the cache table. We use calloc() intentionally, as we 
 		 * need all pointers to be initialized to NULL pointers.
@@ -983,6 +1019,7 @@ CODESTARTparseSelectorAct
 		pData->fileGID = fileGID;
 		pData->dirUID = dirUID;
 		pData->dirGID = dirGID;
+		pData->iZipLevel = iZipLevel;
 
 		/* at this stage, we ignore the return value of prepareFile, this is taken
 		 * care of in later steps. -- rgerhards, 2009-03-19
@@ -1021,6 +1058,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
 	fDirCreateMode = 0700;
 	bCreateDirs = 1;
 	bEnableSync = 0;
+	iZipLevel = 0;
 	if(pszTplName != NULL) {
 		free(pszTplName);
 		pszTplName = NULL;
@@ -1046,6 +1084,8 @@ ENDdoHUP
 
 BEGINmodExit
 CODESTARTmodExit
+	objRelease(errmsg, CORE_COMPONENT);
+	objRelease(strm, CORE_COMPONENT);
 	objRelease(zlibw, LM_ZLIBW_FILENAME);
 	free(pszTplName);
 ENDmodExit
@@ -1062,9 +1102,11 @@ BEGINmodInit(File)
 CODESTARTmodInit
 	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
 CODEmodInit_QueryRegCFSLineHdlr
-	CHKiRet(objUse(zlibw, LM_ZLIBW_FILENAME));
 	CHKiRet(objUse(errmsg, CORE_COMPONENT));
+	CHKiRet(objUse(zlibw, LM_ZLIBW_FILENAME));
+	CHKiRet(objUse(strm, CORE_COMPONENT));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"dynafilecachesize", 0, eCmdHdlrInt, (void*) setDynaFileCacheSize, NULL, STD_LOADABLE_MODULE_ID));
+	CHKiRet(omsdRegCFSLineHdlr((uchar *)"omfileziplevel", 0, eCmdHdlrInt, NULL, &iZipLevel, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"dirowner", 0, eCmdHdlrUID, NULL, &dirUID, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"dirgroup", 0, eCmdHdlrGID, NULL, &dirGID, STD_LOADABLE_MODULE_ID));
 	CHKiRet(omsdRegCFSLineHdlr((uchar *)"fileowner", 0, eCmdHdlrUID, NULL, &fileUID, STD_LOADABLE_MODULE_ID));