Merge branch 'master' of git://anonscm.debian.org/collab-maint/rsyslog

Conflicts:
	debian/changelog
	debian/patches/series
	debian/rules

1 files changed, 343 insertions, 145 deletions

diff --git a/tools/omfwd.c b/tools/omfwd.c
index 129392d..538281d 100644
--- a/tools/omfwd.c
+++ b/tools/omfwd.c
@@ -4,7 +4,7 @@
  * NOTE: read comments in module-template.h to understand how this file
  *       works!
  *
- * Copyright 2007-2012 Adiscon GmbH.
+ * Copyright 2007-2013 Adiscon GmbH.
  *
  * This file is part of rsyslog.
  *
@@ -21,9 +21,6 @@
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
- *
- * TODO v6 config:
- * - permitted peer *list*
  */
 #include "config.h"
 #include "rsyslog.h"
@@ -39,6 +36,7 @@
 #include <errno.h>
 #include <ctype.h>
 #include <unistd.h>
+#include <stdint.h>
 #ifdef USE_NETZIP
 #include <zlib.h>
 #endif
@@ -74,32 +72,51 @@ DEFobjCurrIf(netstrms)
 DEFobjCurrIf(netstrm)
 DEFobjCurrIf(tcpclt)
 
+
+/* some local constants (just) for better readybility */
+#define IS_FLUSH 1
+#define NO_FLUSH 0
+
 typedef struct _instanceData {
 	uchar 	*tplName;	/* name of assigned template */
-	netstrms_t *pNS; /* netstream subsystem */
-	netstrm_t *pNetstrm; /* our output netstream */
 	uchar *pszStrmDrvr;
 	uchar *pszStrmDrvrAuthMode;
 	permittedPeers_t *pPermPeers;
 	int iStrmDrvrMode;
 	char	*target;
-	int *pSockArray;	/* sockets to use for UDP */
-	int bIsConnected;  /* are we connected to remote host? 0 - no, 1 - yes, UDP means addr resolved */
-	struct addrinfo *f_addr;
 	int compressionLevel;	/* 0 - no compression, else level for zlib */
 	char *port;
 	int protocol;
 	int iRebindInterval;	/* rebind interval */
-	int nXmit;		/* number of transmissions since last (re-)bind */
 #	define	FORW_UDP 0
 #	define	FORW_TCP 1
 	/* following fields for TCP-based delivery */
 	TCPFRAMINGMODE tcp_framing;
 	int bResendLastOnRecon; /* should the last message be re-sent on a successful reconnect? */
+#	define COMPRESS_NEVER 0
+#	define COMPRESS_SINGLE_MSG 1	/* old, single-message compression */
+	/* all other settings are for stream-compression */
+#	define COMPRESS_STREAM_ALWAYS 2
+	uint8_t compressionMode;
+	int errsToReport;	/* max number of errors to report (per instance) */
+	sbool strmCompFlushOnTxEnd; /* flush stream compression on transaction end? */
+} instanceData;
+
+typedef struct wrkrInstanceData {
+	instanceData *pData;
+	netstrms_t *pNS; /* netstream subsystem */
+	netstrm_t *pNetstrm; /* our output netstream */
+	struct addrinfo *f_addr;
+	int *pSockArray;	/* sockets to use for UDP */
+	int bIsConnected;  /* are we connected to remote host? 0 - no, 1 - yes, UDP means addr resolved */
+	int nXmit;		/* number of transmissions since last (re-)bind */
 	tcpclt_t *pTCPClt;	/* our tcpclt object */
+	sbool bzInitDone; /* did we do an init of zstrm already? */
+	z_stream zstrm;	/* zip stream to use for tcp compression */
 	uchar sndBuf[16*1024];	/* this is intensionally fixed -- see no good reason to make configurable */
 	unsigned offsSndBuf;	/* next free spot in send buffer */
-} instanceData;
+	int errsToReport;	/* (remaining) number of errors to report */
+} wrkrInstanceData_t;
 
 /* config data */
 typedef struct configSettings_s {
@@ -132,6 +149,9 @@ static struct cnfparamdescr actpdescr[] = {
 	{ "protocol", eCmdHdlrGetWord, 0 },
 	{ "tcp_framing", eCmdHdlrGetWord, 0 },
 	{ "ziplevel", eCmdHdlrInt, 0 },
+	{ "compression.mode", eCmdHdlrGetWord, 0 },
+	{ "compression.stream.flushontxend", eCmdHdlrBinary, 0 },
+	{ "maxerrormessages", eCmdHdlrInt, 0 },
 	{ "rebindinterval", eCmdHdlrInt, 0 },
 	{ "streamdriver", eCmdHdlrGetWord, 0 },
 	{ "streamdrivermode", eCmdHdlrInt, 0 },
@@ -155,6 +175,9 @@ static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current l
 static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current exec process */
 
 
+static rsRetVal initTCP(wrkrInstanceData_t *pWrkrData);
+
+
 BEGINinitConfVars		/* (re)set config variables to default values */
 CODESTARTinitConfVars 
 	cs.pszTplName = NULL; /* name of the default template to use */
@@ -168,7 +191,8 @@ CODESTARTinitConfVars
 ENDinitConfVars
 
 
-static rsRetVal doTryResume(instanceData *pData);
+static rsRetVal doTryResume(wrkrInstanceData_t *);
+static rsRetVal doZipFinish(wrkrInstanceData_t *);
 
 /* this function gets the default template. It coordinates action between
  * old-style and new-style configuration parts.
@@ -212,17 +236,16 @@ finalize_it:
  * rgerhards, 2009-05-29
  */
 static rsRetVal
-closeUDPSockets(instanceData *pData)
+closeUDPSockets(wrkrInstanceData_t *pWrkrData)
 {
 	DEFiRet;
-	assert(pData != NULL);
-	if(pData->pSockArray != NULL) {
-		net.closeUDPListenSockets(pData->pSockArray);
-		pData->pSockArray = NULL;
-		freeaddrinfo(pData->f_addr);
-		pData->f_addr = NULL;
+	if(pWrkrData->pSockArray != NULL) {
+		net.closeUDPListenSockets(pWrkrData->pSockArray);
+		pWrkrData->pSockArray = NULL;
+		freeaddrinfo(pWrkrData->f_addr);
+		pWrkrData->f_addr = NULL;
 	}
-pData->bIsConnected = 0; // TODO: remove this variable altogether
+pWrkrData->bIsConnected = 0; // TODO: remove this variable altogether
 	RETiRet;
 }
 
@@ -233,17 +256,17 @@ pData->bIsConnected = 0; // TODO: remove this variable altogether
  * rgerhards, 2008-06-04
  * Note that we DO NOT discard the current buffer contents
  * (if any). This permits us to save data between sessions. In
- * the wort case, some duplication occurs, but we do not
+ * the worst case, some duplication occurs, but we do not
  * loose data.
  */
 static inline void
-DestructTCPInstanceData(instanceData *pData)
+DestructTCPInstanceData(wrkrInstanceData_t *pWrkrData)
 {
-	assert(pData != NULL);
-	if(pData->pNetstrm != NULL)
-		netstrm.Destruct(&pData->pNetstrm);
-	if(pData->pNS != NULL)
-		netstrms.Destruct(&pData->pNS);
+	doZipFinish(pWrkrData);
+	if(pWrkrData->pNetstrm != NULL)
+		netstrm.Destruct(&pWrkrData->pNetstrm);
+	if(pWrkrData->pNS != NULL)
+		netstrms.Destruct(&pWrkrData->pNS);
 }
 
 
@@ -314,10 +337,25 @@ ENDfreeCnf
 
 BEGINcreateInstance
 CODESTARTcreateInstance
-	pData->offsSndBuf = 0;
+	pData->errsToReport = 5;
+	if(cs.pszStrmDrvr != NULL)
+		CHKmalloc(pData->pszStrmDrvr = (uchar*)strdup((char*)cs.pszStrmDrvr));
+	if(cs.pszStrmDrvrAuthMode != NULL)
+		CHKmalloc(pData->pszStrmDrvrAuthMode =
+				     (uchar*)strdup((char*)cs.pszStrmDrvrAuthMode));
+finalize_it:
 ENDcreateInstance
 
 
+BEGINcreateWrkrInstance
+CODESTARTcreateWrkrInstance
+	dbgprintf("DDDD: createWrkrInstance: pWrkrData %p\n", pWrkrData);
+	pWrkrData->offsSndBuf = 0;
+	pWrkrData->errsToReport = pData->errsToReport;
+	iRet = initTCP(pWrkrData);
+ENDcreateWrkrInstance
+
+
 BEGINisCompatibleWithFeature
 CODESTARTisCompatibleWithFeature
 	if(eFeat == sFEATURERepeatedMsgReduction)
@@ -327,22 +365,25 @@ ENDisCompatibleWithFeature
 
 BEGINfreeInstance
 CODESTARTfreeInstance
-	/* final cleanup */
-	DestructTCPInstanceData(pData);
-	closeUDPSockets(pData);
-
-	if(pData->protocol == FORW_TCP) {
-		tcpclt.Destruct(&pData->pTCPClt);
-	}
-
-	free(pData->port);
-	free(pData->target);
 	free(pData->pszStrmDrvr);
 	free(pData->pszStrmDrvrAuthMode);
+	free(pData->port);
+	free(pData->target);
 	net.DestructPermittedPeers(&pData->pPermPeers);
 ENDfreeInstance
 
 
+BEGINfreeWrkrInstance
+CODESTARTfreeWrkrInstance
+	DestructTCPInstanceData(pWrkrData);
+	closeUDPSockets(pWrkrData);
+
+	if(pWrkrData->pData->protocol == FORW_TCP) {
+		tcpclt.Destruct(&pWrkrData->pTCPClt);
+	}
+ENDfreeWrkrInstance
+
+
 BEGINdbgPrintInstInfo
 CODESTARTdbgPrintInstInfo
 	dbgprintf("%s", pData->target);
@@ -352,25 +393,29 @@ ENDdbgPrintInstInfo
 /* Send a message via UDP
  * rgehards, 2007-12-20
  */
-static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
+static rsRetVal UDPSend(wrkrInstanceData_t *__restrict__ const pWrkrData,
+	uchar *__restrict__ const msg,
+	const size_t len)
 {
 	DEFiRet;
 	struct addrinfo *r;
 	int i;
 	unsigned lsent = 0;
-	int bSendSuccess;
+	sbool bSendSuccess;
+	int lasterrno = ENOENT;
+	char errStr[1024];
 
-	if(pData->iRebindInterval && (pData->nXmit++ % pData->iRebindInterval == 0)) {
+	if(pWrkrData->pData->iRebindInterval && (pWrkrData->nXmit++ % pWrkrData->pData->iRebindInterval == 0)) {
 		dbgprintf("omfwd dropping UDP 'connection' (as configured)\n");
-		pData->nXmit = 1;	/* else we have an addtl wrap at 2^31-1 */
-		CHKiRet(closeUDPSockets(pData));
+		pWrkrData->nXmit = 1;	/* else we have an addtl wrap at 2^31-1 */
+		CHKiRet(closeUDPSockets(pWrkrData));
 	}
 
-	if(pData->pSockArray == NULL) {
-		CHKiRet(doTryResume(pData));
+	if(pWrkrData->pSockArray == NULL) {
+		CHKiRet(doTryResume(pWrkrData));
 	}
 
-	if(pData->pSockArray != NULL) {
+	if(pWrkrData->pSockArray != NULL) {
 		/* we need to track if we have success sending to the remote
 		 * peer. Success is indicated by at least one sendto() call
 		 * succeeding. We track this be bSendSuccess. We can not simply
@@ -379,25 +424,37 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
 		 * the sendto() succeeded. -- rgerhards, 2007-06-22
 		 */
 		bSendSuccess = RSFALSE;
-		for (r = pData->f_addr; r; r = r->ai_next) {
-			for (i = 0; i < *pData->pSockArray; i++) {
-			       lsent = sendto(pData->pSockArray[i+1], msg, len, 0, r->ai_addr, r->ai_addrlen);
+		for (r = pWrkrData->f_addr; r; r = r->ai_next) {
+			for (i = 0; i < *pWrkrData->pSockArray; i++) {
+			       lsent = sendto(pWrkrData->pSockArray[i+1], msg, len, 0, r->ai_addr, r->ai_addrlen);
 				if (lsent == len) {
 					bSendSuccess = RSTRUE;
 					break;
 				} else {
-					int eno = errno;
-					char errStr[1024];
-					dbgprintf("sendto() error: %d = %s.\n",
-						eno, rs_strerror_r(eno, errStr, sizeof(errStr)));
+					lasterrno = errno;
+					DBGPRINTF("sendto() error: %d = %s.\n",
+						lasterrno,
+						rs_strerror_r(lasterrno, errStr, sizeof(errStr)));
 				}
 			}
 			if (lsent == len && !send_to_all)
 			       break;
 		}
 		/* finished looping */
-		if (bSendSuccess == RSFALSE) {
+		if(bSendSuccess == RSFALSE) {
 			dbgprintf("error forwarding via udp, suspending\n");
+			if(pWrkrData->errsToReport > 0) {
+				rs_strerror_r(lasterrno, errStr, sizeof(errStr));
+				errmsg.LogError(0, RS_RET_ERR_UDPSEND, "omfwd: error sending "
+						"via udp: %s", errStr);
+				if(pWrkrData->errsToReport == 1) {
+					errmsg.LogError(0, RS_RET_LAST_ERRREPORT, "omfwd: "
+							"max number of error message emitted "
+							"- further messages will be "
+							"suppressed");
+				}
+				--pWrkrData->errsToReport;
+			}
 			iRet = RS_RET_SUSPENDED;
 		}
 	}
@@ -423,24 +480,19 @@ finalize_it:
 
 /* CODE FOR SENDING TCP MESSAGES */
 
-
-/* Send a buffer via TCP. Usually, this is used to send the current
- * send buffer, but if a message is larger than the buffer, we need to
- * have the capability to send the message buffer directly.
- * rgerhards, 2011-04-04
- */
 static rsRetVal
-TCPSendBuf(instanceData *pData, uchar *buf, unsigned len)
+TCPSendBufUncompressed(wrkrInstanceData_t *pWrkrData, uchar *buf, unsigned len)
 {
 	DEFiRet;
 	unsigned alreadySent;
 	ssize_t lenSend;
 
 	alreadySent = 0;
-	CHKiRet(netstrm.CheckConnection(pData->pNetstrm)); /* hack for plain tcp syslog - see ptcp driver for details */
+	CHKiRet(netstrm.CheckConnection(pWrkrData->pNetstrm)); /* hack for plain tcp syslog - see ptcp driver for details */
+
 	while(alreadySent != len) {
 		lenSend = len - alreadySent;
-		CHKiRet(netstrm.Send(pData->pNetstrm, buf+alreadySent, &lenSend));
+		CHKiRet(netstrm.Send(pWrkrData->pNetstrm, buf+alreadySent, &lenSend));
 		DBGPRINTF("omfwd: TCP sent %ld bytes, requested %u\n", (long) lenSend, len - alreadySent);
 		alreadySent += lenSend;
 	}
@@ -449,38 +501,136 @@ finalize_it:
 	if(iRet != RS_RET_OK) {
 		/* error! */
 		dbgprintf("TCPSendBuf error %d, destruct TCP Connection!\n", iRet);
-		DestructTCPInstanceData(pData);
+		DestructTCPInstanceData(pWrkrData);
 		iRet = RS_RET_SUSPENDED;
 	}
 	RETiRet;
 }
 
+static rsRetVal
+TCPSendBufCompressed(wrkrInstanceData_t *pWrkrData, uchar *buf, unsigned len, sbool bIsFlush)
+{
+	int zRet;	/* zlib return state */
+	unsigned outavail;
+	uchar zipBuf[32*1024];
+	int op;
+	DEFiRet;
+
+	if(!pWrkrData->bzInitDone) {
+		/* allocate deflate state */
+		pWrkrData->zstrm.zalloc = Z_NULL;
+		pWrkrData->zstrm.zfree = Z_NULL;
+		pWrkrData->zstrm.opaque = Z_NULL;
+		/* see note in file header for the params we use with deflateInit2() */
+		zRet = deflateInit(&pWrkrData->zstrm, 9);
+		if(zRet != Z_OK) {
+			DBGPRINTF("error %d returned from zlib/deflateInit()\n", zRet);
+			ABORT_FINALIZE(RS_RET_ZLIB_ERR);
+		}
+		pWrkrData->bzInitDone = RSTRUE;
+	}
+
+	/* now doing the compression */
+	pWrkrData->zstrm.next_in = (Bytef*) buf;
+	pWrkrData->zstrm.avail_in = len;
+	if(pWrkrData->pData->strmCompFlushOnTxEnd && bIsFlush)
+		op = Z_SYNC_FLUSH;
+	else
+		op = Z_NO_FLUSH;
+	/* run deflate() on buffer until everything has been compressed */
+	do {
+		DBGPRINTF("omfwd: in deflate() loop, avail_in %d, total_in %ld, isFlush %d\n", pWrkrData->zstrm.avail_in, pWrkrData->zstrm.total_in, bIsFlush);
+		pWrkrData->zstrm.avail_out = sizeof(zipBuf);
+		pWrkrData->zstrm.next_out = zipBuf;
+		zRet = deflate(&pWrkrData->zstrm, op);    /* no bad return value */
+		DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pWrkrData->zstrm.avail_out);
+		outavail = sizeof(zipBuf) - pWrkrData->zstrm.avail_out;
+		if(outavail != 0) {
+			CHKiRet(TCPSendBufUncompressed(pWrkrData, zipBuf, outavail));
+		}
+	} while (pWrkrData->zstrm.avail_out == 0);
+
+finalize_it:
+	RETiRet;
+}
+
+static rsRetVal
+TCPSendBuf(wrkrInstanceData_t *pWrkrData, uchar *buf, unsigned len, sbool bIsFlush)
+{
+	DEFiRet;
+	if(pWrkrData->pData->compressionMode >= COMPRESS_STREAM_ALWAYS)
+		iRet = TCPSendBufCompressed(pWrkrData, buf, len, bIsFlush);
+	else
+		iRet = TCPSendBufUncompressed(pWrkrData, buf, len);
+	RETiRet;
+}
+
+/* finish zlib buffer, to be called before closing the ZIP file (if
+ * running in stream mode).
+ */
+static rsRetVal
+doZipFinish(wrkrInstanceData_t *pWrkrData)
+{
+	int zRet;	/* zlib return state */
+	DEFiRet;
+	unsigned outavail;
+	uchar zipBuf[32*1024];
+
+	if(!pWrkrData->bzInitDone)
+		goto done;
+
+// TODO: can we get this into a single common function?
+dbgprintf("DDDD: in doZipFinish()\n");
+	pWrkrData->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", pWrkrData->zstrm.avail_in, pWrkrData->zstrm.total_in);
+		pWrkrData->zstrm.avail_out = sizeof(zipBuf);
+		pWrkrData->zstrm.next_out = zipBuf;
+		zRet = deflate(&pWrkrData->zstrm, Z_FINISH);    /* no bad return value */
+		DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pWrkrData->zstrm.avail_out);
+		outavail = sizeof(zipBuf) - pWrkrData->zstrm.avail_out;
+		if(outavail != 0) {
+			CHKiRet(TCPSendBufUncompressed(pWrkrData, zipBuf, outavail));
+		}
+	} while (pWrkrData->zstrm.avail_out == 0);
+
+finalize_it:
+	zRet = deflateEnd(&pWrkrData->zstrm);
+	if(zRet != Z_OK) {
+		DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet);
+	}
+
+	pWrkrData->bzInitDone = 0;
+done:	RETiRet;
+}
+
 
 /* Add frame to send buffer (or send, if requried)
  */
 static rsRetVal TCPSendFrame(void *pvData, char *msg, size_t len)
 {
 	DEFiRet;
-	instanceData *pData = (instanceData *) pvData;
+	wrkrInstanceData_t *pWrkrData = (wrkrInstanceData_t *) pvData;
 
 	DBGPRINTF("omfwd: add %u bytes to send buffer (curr offs %u)\n",
-		(unsigned) len, pData->offsSndBuf);
-	if(pData->offsSndBuf != 0 && pData->offsSndBuf + len >= sizeof(pData->sndBuf)) {
+		(unsigned) len, pWrkrData->offsSndBuf);
+	if(pWrkrData->offsSndBuf != 0 && pWrkrData->offsSndBuf + len >= sizeof(pWrkrData->sndBuf)) {
 		/* no buffer space left, need to commit previous records */
-		CHKiRet(TCPSendBuf(pData, pData->sndBuf, pData->offsSndBuf));
-		pData->offsSndBuf = 0;
+		CHKiRet(TCPSendBuf(pWrkrData, pWrkrData->sndBuf, pWrkrData->offsSndBuf, NO_FLUSH));
+		pWrkrData->offsSndBuf = 0;
 		iRet = RS_RET_PREVIOUS_COMMITTED;
 	}
 
 	/* check if the message is too large to fit into buffer */
-	if(len > sizeof(pData->sndBuf)) {
-		CHKiRet(TCPSendBuf(pData, (uchar*)msg, len));
+	if(len > sizeof(pWrkrData->sndBuf)) {
+		CHKiRet(TCPSendBuf(pWrkrData, (uchar*)msg, len, NO_FLUSH));
 		ABORT_FINALIZE(RS_RET_OK);	/* committed everything so far */
 	}
 
 	/* we now know the buffer has enough free space */
-	memcpy(pData->sndBuf + pData->offsSndBuf, msg, len);
-	pData->offsSndBuf += len;
+	memcpy(pWrkrData->sndBuf + pWrkrData->offsSndBuf, msg, len);
+	pWrkrData->offsSndBuf += len;
 	iRet = RS_RET_DEFER_COMMIT;
 
 finalize_it:
@@ -495,11 +645,10 @@ finalize_it:
 static rsRetVal TCPSendPrepRetry(void *pvData)
 {
 	DEFiRet;
-	instanceData *pData = (instanceData *) pvData;
-dbgprintf("TCPSendPrepRetry performs a DestructTCPInstanceData\n");
+	wrkrInstanceData_t *pWrkrData = (wrkrInstanceData_t *) pvData;
 
-	assert(pData != NULL);
-	DestructTCPInstanceData(pData);
+	assert(pWrkrData != NULL);
+	DestructTCPInstanceData(pWrkrData);
 	RETiRet;
 }
 
@@ -510,36 +659,39 @@ dbgprintf("TCPSendPrepRetry performs a DestructTCPInstanceData\n");
 static rsRetVal TCPSendInit(void *pvData)
 {
 	DEFiRet;
-	instanceData *pData = (instanceData *) pvData;
+	wrkrInstanceData_t *pWrkrData = (wrkrInstanceData_t *) pvData;
+	instanceData *pData;
 
-	assert(pData != NULL);
-	if(pData->pNetstrm == NULL) {
+	assert(pWrkrData != NULL);
+	pData = pWrkrData->pData;
+
+	if(pWrkrData->pNetstrm == NULL) {
 		dbgprintf("TCPSendInit CREATE\n");
-		CHKiRet(netstrms.Construct(&pData->pNS));
+		CHKiRet(netstrms.Construct(&pWrkrData->pNS));
 		/* the stream driver must be set before the object is finalized! */
-		CHKiRet(netstrms.SetDrvrName(pData->pNS, pData->pszStrmDrvr));
-		CHKiRet(netstrms.ConstructFinalize(pData->pNS));
+		CHKiRet(netstrms.SetDrvrName(pWrkrData->pNS, pData->pszStrmDrvr));
+		CHKiRet(netstrms.ConstructFinalize(pWrkrData->pNS));
 
 		/* now create the actual stream and connect to the server */
-		CHKiRet(netstrms.CreateStrm(pData->pNS, &pData->pNetstrm));
-		CHKiRet(netstrm.ConstructFinalize(pData->pNetstrm));
-		CHKiRet(netstrm.SetDrvrMode(pData->pNetstrm, pData->iStrmDrvrMode));
+		CHKiRet(netstrms.CreateStrm(pWrkrData->pNS, &pWrkrData->pNetstrm));
+		CHKiRet(netstrm.ConstructFinalize(pWrkrData->pNetstrm));
+		CHKiRet(netstrm.SetDrvrMode(pWrkrData->pNetstrm, pData->iStrmDrvrMode));
 		/* now set optional params, but only if they were actually configured */
 		if(pData->pszStrmDrvrAuthMode != NULL) {
-			CHKiRet(netstrm.SetDrvrAuthMode(pData->pNetstrm, pData->pszStrmDrvrAuthMode));
+			CHKiRet(netstrm.SetDrvrAuthMode(pWrkrData->pNetstrm, pData->pszStrmDrvrAuthMode));
 		}
 		if(pData->pPermPeers != NULL) {
-			CHKiRet(netstrm.SetDrvrPermPeers(pData->pNetstrm, pData->pPermPeers));
+			CHKiRet(netstrm.SetDrvrPermPeers(pWrkrData->pNetstrm, pData->pPermPeers));
 		}
 		/* params set, now connect */
-		CHKiRet(netstrm.Connect(pData->pNetstrm, glbl.GetDefPFFamily(),
+		CHKiRet(netstrm.Connect(pWrkrData->pNetstrm, glbl.GetDefPFFamily(),
 			(uchar*)pData->port, (uchar*)pData->target));
 	}
 
 finalize_it:
 	if(iRet != RS_RET_OK) {
 		dbgprintf("TCPSendInit FAILED with %d.\n", iRet);
-		DestructTCPInstanceData(pData);
+		DestructTCPInstanceData(pWrkrData);
 	}
 
 	RETiRet;
@@ -549,15 +701,17 @@ finalize_it:
 /* try to resume connection if it is not ready
  * rgerhards, 2007-08-02
  */
-static rsRetVal doTryResume(instanceData *pData)
+static rsRetVal doTryResume(wrkrInstanceData_t *pWrkrData)
 {
 	int iErr;
 	struct addrinfo *res;
 	struct addrinfo hints;
+	instanceData *pData;
 	DEFiRet;
 
-	if(pData->bIsConnected)
+	if(pWrkrData->bIsConnected)
 		FINALIZE;
+	pData = pWrkrData->pData;
 
 	/* The remote address is not yet known and needs to be obtained */
 	dbgprintf(" %s\n", pData->target);
@@ -573,20 +727,20 @@ static rsRetVal doTryResume(instanceData *pData)
 			ABORT_FINALIZE(RS_RET_SUSPENDED);
 		}
 		dbgprintf("%s found, resuming.\n", pData->target);
-		pData->f_addr = res;
-		pData->bIsConnected = 1;
-		if(pData->pSockArray == NULL) {
-			pData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0);
+		pWrkrData->f_addr = res;
+		pWrkrData->bIsConnected = 1;
+		if(pWrkrData->pSockArray == NULL) {
+			pWrkrData->pSockArray = net.create_udp_socket((uchar*)pData->target, NULL, 0, 0);
 		}
 	} else {
-		CHKiRet(TCPSendInit((void*)pData));
+		CHKiRet(TCPSendInit((void*)pWrkrData));
 	}
 
 finalize_it:
 	if(iRet != RS_RET_OK) {
-		if(pData->f_addr != NULL) {
-			freeaddrinfo(pData->f_addr);
-			pData->f_addr = NULL;
+		if(pWrkrData->f_addr != NULL) {
+			freeaddrinfo(pWrkrData->f_addr);
+			pWrkrData->f_addr = NULL;
 		}
 		iRet = RS_RET_SUSPENDED;
 	}
@@ -597,33 +751,35 @@ finalize_it:
 
 BEGINtryResume
 CODESTARTtryResume
-	iRet = doTryResume(pData);
+	dbgprintf("DDDD: tryResume: pWrkrData %p\n", pWrkrData);
+	iRet = doTryResume(pWrkrData);
 ENDtryResume
 
 
 BEGINbeginTransaction
 CODESTARTbeginTransaction
 dbgprintf("omfwd: beginTransaction\n");
+	iRet = doTryResume(pWrkrData);
 ENDbeginTransaction
 
 
-BEGINdoAction
-	char *psz; /* temporary buffering */
+static rsRetVal
+processMsg(wrkrInstanceData_t *__restrict__ const pWrkrData,
+	actWrkrIParams_t *__restrict__ const iparam)
+{
+	uchar *psz; /* temporary buffering */
 	register unsigned l;
 	int iMaxLine;
 #	ifdef	USE_NETZIP
 	Bytef *out = NULL; /* for compression */
 #	endif
-CODESTARTdoAction
-	CHKiRet(doTryResume(pData));
+	instanceData *__restrict__ const pData = pWrkrData->pData;
+	DEFiRet;
 
 	iMaxLine = glbl.GetMaxLine();
 
-	dbgprintf(" %s:%s/%s\n", pData->target, pData->port,
-		 pData->protocol == FORW_UDP ? "udp" : "tcp");
-
-	psz = (char*) ppString[0];
-	l = strlen((char*) psz);
+	psz = iparam->param;
+	l = iparam->lenStr;
 	if((int) l > iMaxLine)
 		l = iMaxLine;
 
@@ -636,11 +792,10 @@ CODESTARTdoAction
 	 * hard-coded but this may be changed to a config parameter.
 	 * rgerhards, 2006-11-30
 	 */
-	if(pData->compressionLevel && (l > CONF_MIN_SIZE_FOR_COMPRESS)) {
+	if(pData->compressionMode == COMPRESS_SINGLE_MSG && (l > CONF_MIN_SIZE_FOR_COMPRESS)) {
 		uLongf destLen = iMaxLine + iMaxLine/100 +12; /* recommended value from zlib doc */
 		uLong srcLen = l;
 		int ret;
-		/* TODO: optimize malloc sequence? -- rgerhards, 2008-09-02 */
 		CHKmalloc(out = (Bytef*) MALLOC(destLen));
 		out[0] = 'z';
 		out[1] = '\0';
@@ -660,7 +815,7 @@ CODESTARTdoAction
 		} else if(destLen+1 < l) {
 			/* only use compression if there is a gain in using it! */
 			dbgprintf("there is gain in compression, so we do it\n");
-			psz = (char*) out;
+			psz = out;
 			l = destLen + 1; /* take care for the "z" at message start! */
 		}
 		++destLen;
@@ -669,14 +824,14 @@ CODESTARTdoAction
 
 	if(pData->protocol == FORW_UDP) {
 		/* forward via UDP */
-		CHKiRet(UDPSend(pData, psz, l));
+		CHKiRet(UDPSend(pWrkrData, psz, l));
 	} else {
 		/* forward via TCP */
-		iRet = tcpclt.Send(pData->pTCPClt, pData, psz, l);
+		iRet = tcpclt.Send(pWrkrData->pTCPClt, pWrkrData, (char *)psz, l);
 		if(iRet != RS_RET_OK && iRet != RS_RET_DEFER_COMMIT && iRet != RS_RET_PREVIOUS_COMMITTED) {
 			/* error! */
 			dbgprintf("error forwarding via tcp, suspending\n");
-			DestructTCPInstanceData(pData);
+			DestructTCPInstanceData(pWrkrData);
 			iRet = RS_RET_SUSPENDED;
 		}
 	}
@@ -684,18 +839,30 @@ finalize_it:
 #	ifdef USE_NETZIP
 	free(out); /* is NULL if it was never used... */
 #	endif
-ENDdoAction
+	RETiRet;
+}
+
+BEGINcommitTransaction
+	unsigned i;
+CODESTARTcommitTransaction
+	CHKiRet(doTryResume(pWrkrData));
 
+	dbgprintf(" %s:%s/%s\n", pWrkrData->pData->target, pWrkrData->pData->port,
+		 pWrkrData->pData->protocol == FORW_UDP ? "udp" : "tcp");
 
-BEGINendTransaction
-CODESTARTendTransaction
-dbgprintf("omfwd: endTransaction, offsSndBuf %u\n", pData->offsSndBuf);
-	if(pData->offsSndBuf != 0) {
-		iRet = TCPSendBuf(pData, pData->sndBuf, pData->offsSndBuf);
-		pData->offsSndBuf = 0;
+	for(i = 0 ; i < nParams ; ++i) {
+		iRet = processMsg(pWrkrData, &actParam(pParams, 1, i, 0));
+		if(iRet != RS_RET_OK && iRet != RS_RET_DEFER_COMMIT && iRet != RS_RET_PREVIOUS_COMMITTED)
+			FINALIZE;
 	}
-ENDendTransaction
 
+dbgprintf("omfwd: endTransaction, offsSndBuf %u, iRet %d\n", pWrkrData->offsSndBuf, iRet);
+	if(pWrkrData->offsSndBuf != 0) {
+		iRet = TCPSendBuf(pWrkrData, pWrkrData->sndBuf, pWrkrData->offsSndBuf, IS_FLUSH);
+		pWrkrData->offsSndBuf = 0;
+	}
+finalize_it:
+ENDcommitTransaction
 
 
 /* This function loads TCP support, if not already loaded. It will be called
@@ -719,25 +886,22 @@ finalize_it:
  * created.
  */
 static rsRetVal
-initTCP(instanceData *pData)
+initTCP(wrkrInstanceData_t *pWrkrData)
 {
+	instanceData *pData;
 	DEFiRet;
+
+	pData = pWrkrData->pData;
 	if(pData->protocol == FORW_TCP) {
 		/* create our tcpclt */
-		CHKiRet(tcpclt.Construct(&pData->pTCPClt));
-		CHKiRet(tcpclt.SetResendLastOnRecon(pData->pTCPClt, pData->bResendLastOnRecon));
+		CHKiRet(tcpclt.Construct(&pWrkrData->pTCPClt));
+		CHKiRet(tcpclt.SetResendLastOnRecon(pWrkrData->pTCPClt, pData->bResendLastOnRecon));
 		/* and set callbacks */
-		CHKiRet(tcpclt.SetSendInit(pData->pTCPClt, TCPSendInit));
-		CHKiRet(tcpclt.SetSendFrame(pData->pTCPClt, TCPSendFrame));
-		CHKiRet(tcpclt.SetSendPrepRetry(pData->pTCPClt, TCPSendPrepRetry));
-		CHKiRet(tcpclt.SetFraming(pData->pTCPClt, pData->tcp_framing));
-		CHKiRet(tcpclt.SetRebindInterval(pData->pTCPClt, pData->iRebindInterval));
-		pData->iStrmDrvrMode = cs.iStrmDrvrMode;
-		if(cs.pszStrmDrvr != NULL)
-			CHKmalloc(pData->pszStrmDrvr = (uchar*)strdup((char*)cs.pszStrmDrvr));
-		if(cs.pszStrmDrvrAuthMode != NULL)
-			CHKmalloc(pData->pszStrmDrvrAuthMode =
-				     (uchar*)strdup((char*)cs.pszStrmDrvrAuthMode));
+		CHKiRet(tcpclt.SetSendInit(pWrkrData->pTCPClt, TCPSendInit));
+		CHKiRet(tcpclt.SetSendFrame(pWrkrData->pTCPClt, TCPSendFrame));
+		CHKiRet(tcpclt.SetSendPrepRetry(pWrkrData->pTCPClt, TCPSendPrepRetry));
+		CHKiRet(tcpclt.SetFraming(pWrkrData->pTCPClt, pData->tcp_framing));
+		CHKiRet(tcpclt.SetRebindInterval(pWrkrData->pTCPClt, pData->iRebindInterval));
 	}
 finalize_it:
 	RETiRet;
@@ -756,14 +920,19 @@ setInstParamDefaults(instanceData *pData)
 	pData->iRebindInterval = 0;
 	pData->bResendLastOnRecon = 0; 
 	pData->pPermPeers = NULL;
-	pData->compressionLevel = 0;
+	pData->compressionLevel = 9;
+	pData->strmCompFlushOnTxEnd = 1;
+	pData->compressionMode = COMPRESS_NEVER;
+	pData->errsToReport = 5;
 }
 
 BEGINnewActInst
 	struct cnfparamvals *pvals;
 	uchar *tplToUse;
+	char *cstr;
 	int i;
 	rsRetVal localRet;
+	int complevel = -1;
 CODESTARTnewActInst
 	DBGPRINTF("newActInst (omfwd)\n");
 
@@ -860,9 +1029,10 @@ CODESTARTnewActInst
 			free(str);
 		} else if(!strcmp(actpblk.descr[i].name, "ziplevel")) {
 #			ifdef USE_NETZIP
-			int complevel = pvals[i].val.d.n;
+			complevel = pvals[i].val.d.n;
 			if(complevel >= 0 && complevel <= 10) {
 				pData->compressionLevel = complevel;
+				pData->compressionMode = COMPRESS_SINGLE_MSG;
 			} else {
 				errmsg.LogError(0, NO_ERRCODE, "Invalid ziplevel %d specified in "
 					 "forwardig action - NOT turning on compression.",
@@ -872,21 +1042,50 @@ CODESTARTnewActInst
 			errmsg.LogError(0, NO_ERRCODE, "Compression requested, but rsyslogd is not compiled "
 				 "with compression support - request ignored.");
 #			endif /* #ifdef USE_NETZIP */
+		} else if(!strcmp(actpblk.descr[i].name, "maxerrormessages")) {
+			pData->errsToReport = (int) pvals[i].val.d.n;
 		} else if(!strcmp(actpblk.descr[i].name, "resendlastmsgonreconnect")) {
 			pData->bResendLastOnRecon = (int) pvals[i].val.d.n;
 		} else if(!strcmp(actpblk.descr[i].name, "template")) {
 			pData->tplName = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL);
+		} else if(!strcmp(actpblk.descr[i].name, "compression.stream.flushontxend")) {
+			pData->strmCompFlushOnTxEnd = (sbool) pvals[i].val.d.n;
+		} else if(!strcmp(actpblk.descr[i].name, "compression.mode")) {
+			cstr = es_str2cstr(pvals[i].val.d.estr, NULL);
+			if(!strcasecmp(cstr, "stream:always")) {
+				pData->compressionMode = COMPRESS_STREAM_ALWAYS;
+			} else if(!strcasecmp(cstr, "none")) {
+				pData->compressionMode = COMPRESS_NEVER;
+			} else if(!strcasecmp(cstr, "single")) {
+				pData->compressionMode = COMPRESS_SINGLE_MSG;
+			} else {
+				errmsg.LogError(0, RS_RET_PARAM_ERROR, "omfwd: invalid value for 'compression.mode' "
+					 "parameter (given is '%s')", cstr);
+				free(cstr);
+				ABORT_FINALIZE(RS_RET_PARAM_ERROR);
+			}
+			free(cstr);
 		} else {
 			DBGPRINTF("omfwd: program error, non-handled "
 			  "param '%s'\n", actpblk.descr[i].name);
 		}
 	}
+
+	if(complevel != -1) {
+		pData->compressionLevel = complevel;
+		if(pData->compressionMode == COMPRESS_NEVER) {
+			/* to keep compatible with pre-7.3.11, only setting the
+			 * compresion level means old-style single-message mode.
+			 */
+			pData->compressionMode = COMPRESS_SINGLE_MSG;
+		}
+	}
+
 	CODE_STD_STRING_REQUESTnewActInst(1)
 
 	tplToUse = ustrdup((pData->tplName == NULL) ? getDfltTpl() : pData->tplName);
 	CHKiRet(OMSRsetEntry(*ppOMSR, 0, tplToUse, OMSR_NO_RQD_TPL_OPTS));
 
-	CHKiRet(initTCP(pData));
 CODE_STD_FINALIZERnewActInst
 	cnfparamvalsDestruct(pvals, &actpblk);
 ENDnewActInst
@@ -948,6 +1147,7 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
 					iLevel = *p - '0';
 					++p; /* eat */
 					pData->compressionLevel = iLevel;
+					pData->compressionMode = COMPRESS_SINGLE_MSG;
 				} else {
 					errmsg.LogError(0, NO_ERRCODE, "Invalid compression level '%c' specified in "
 						 "forwardig action - NOT turning on compression.",
@@ -1025,7 +1225,6 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
 	while(*p && *p != ';'  && *p != '#' && !isspace((int) *p))
 		++p; /*JUST SKIP*/
 
-	/* TODO: make this if go away! */
 	if(*p == ';' || *p == '#' || isspace(*p)) {
 		uchar cTmp = *p;
 		*p = '\0'; /* trick to obtain hostname (later)! */
@@ -1055,7 +1254,6 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
 			cs.pPermPeers = NULL;
 		}
 	}
-	CHKiRet(initTCP(pData));
 CODE_STD_FINALIZERparseSelectorAct
 ENDparseSelectorAct
 
@@ -1090,11 +1288,11 @@ ENDmodExit
 
 BEGINqueryEtryPt
 CODESTARTqueryEtryPt
-CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_STD_OMODTX_QUERIES
+CODEqueryEtryPt_STD_OMOD8_QUERIES
 CODEqueryEtryPt_STD_CONF2_QUERIES
 CODEqueryEtryPt_STD_CONF2_setModCnf_QUERIES
 CODEqueryEtryPt_STD_CONF2_OMOD_QUERIES
-CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */
 ENDqueryEtryPt