diff options
| -rw-r--r-- | plugins/imptcp/imptcp.c | 179 | ||||
| -rw-r--r-- | tools/omfwd.c | 154 |
2 files changed, 308 insertions, 25 deletions
diff --git a/plugins/imptcp/imptcp.c b/plugins/imptcp/imptcp.c index 5c8bb67a..e9a20c1c 100644 --- a/plugins/imptcp/imptcp.c +++ b/plugins/imptcp/imptcp.c @@ -10,7 +10,7 @@ * * File begun on 2010-08-10 by RGerhards * - * Copyright 2007-2012 Rainer Gerhards and Adiscon GmbH. + * Copyright 2007-2013 Rainer Gerhards and Adiscon GmbH. * * This file is part of rsyslog. * @@ -50,6 +50,8 @@ #include <sys/socket.h> #include <sys/epoll.h> #include <netinet/tcp.h> +#include <stdint.h> +#include <zlib.h> #if HAVE_FCNTL_H #include <fcntl.h> #endif @@ -93,6 +95,11 @@ static void * wrkr(void *myself); #define DFLT_wrkrMax 2 +#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 + /* config settings */ typedef struct configSettings_s { int bKeepAlive; /* support keep-alive packets */ @@ -117,6 +124,7 @@ struct instanceConf_s { int bEmitMsgOnClose; int bSuppOctetFram; /* support octet-counted framing? */ int iAddtlFrameDelim; + uint8_t compressionMode; uchar *pszBindPort; /* port to bind to */ uchar *pszBindAddr; /* IP to bind socket to */ uchar *pszBindRuleset; /* name of ruleset to bind to */ @@ -156,6 +164,7 @@ static struct cnfparamdescr inppdescr[] = { { "ruleset", eCmdHdlrString, 0 }, { "supportoctetcountedframing", eCmdHdlrBinary, 0 }, { "notifyonconnectionclose", eCmdHdlrBinary, 0 }, + { "compression.mode", eCmdHdlrGetWord, 0 }, { "keepalive", eCmdHdlrBinary, 0 }, { "keepalive.probes", eCmdHdlrInt, 0 }, { "keepalive.time", eCmdHdlrInt, 0 }, @@ -191,6 +200,7 @@ struct ptcpsrv_s { int iKeepAliveIntvl; int iKeepAliveProbes; int iKeepAliveTime; + uint8_t compressionMode; uchar *pszInputName; prop_t *pInputName; /* InputName in (fast to process) property format */ ruleset_t *pRuleset; @@ -207,11 +217,13 @@ struct ptcpsrv_s { * includes support for doubly-linked list. */ struct ptcpsess_s { -// ptcpsrv_t *pSrv; /* our server TODO: check remove! */ ptcplstn_t *pLstn; /* our listener */ ptcpsess_t *prev, *next; int sock; epolld_t *epd; + sbool bzInitDone; /* did we do an init of zstrm already? */ + z_stream zstrm; /* zip stream to use for tcp compression */ + uint8_t compressionMode; //--- from tcps_sess.h int iMsg; /* index of next char to store in msg */ int bAtStrtOfFram; /* are we at the very beginning of a new frame? */ @@ -239,6 +251,8 @@ struct ptcplstn_s { sbool bSuppOctetFram; epolld_t *epd; statsobj_t *stats; /* listener stats */ + intctr_t rcvdBytes; + intctr_t rcvdDecompressed; STATSCOUNTER_DEF(ctrSubmit, mutCtrSubmit) }; @@ -806,19 +820,19 @@ processDataRcvd(ptcpsess_t *pThis, char c, struct syslogTime *stTime, time_t ttG * EXTRACT from tcps_sess.c */ static rsRetVal -DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen) +DataRcvdUncompressed(ptcpsess_t *pThis, char *pData, size_t iLen, time_t ttGenTime) { multi_submit_t multiSub; msg_t *pMsgs[CONF_NUM_MULTISUB]; struct syslogTime stTime; - time_t ttGenTime; char *pEnd; DEFiRet; assert(pData != NULL); assert(iLen > 0); - datetime.getCurrTime(&stTime, &ttGenTime); + if(ttGenTime == 0) + datetime.getCurrTime(&stTime, &ttGenTime); multiSub.ppMsgs = pMsgs; multiSub.maxElem = CONF_NUM_MULTISUB; multiSub.nElem = 0; @@ -836,6 +850,71 @@ finalize_it: RETiRet; } +static rsRetVal +DataRcvdCompressed(ptcpsess_t *pThis, char *buf, size_t len) +{ + struct syslogTime stTime; + time_t ttGenTime; + int zRet; /* zlib return state */ + unsigned outavail; + uchar zipBuf[64*1024]; // TODO: alloc on heap, and much larger (512KiB? batch size!) + DEFiRet; + // TODO: can we do stats counters? Even if they are not 100% correct under all cases, + // by simply updating the input and output sizes? + uint64_t outtotal; + + assert(iLen > 0); + + datetime.getCurrTime(&stTime, &ttGenTime); + outtotal = 0; + + if(!pThis->bzInitDone) { + /* allocate deflate state */ + pThis->zstrm.zalloc = Z_NULL; + pThis->zstrm.zfree = Z_NULL; + pThis->zstrm.opaque = Z_NULL; + zRet = inflateInit(&pThis->zstrm); + if(zRet != Z_OK) { + DBGPRINTF("imptcp: error %d returned from zlib/inflateInit()\n", zRet); + ABORT_FINALIZE(RS_RET_ZLIB_ERR); + } + pThis->bzInitDone = RSTRUE; + } + + pThis->zstrm.next_in = (Bytef*) buf; + pThis->zstrm.avail_in = len; + /* run inflate() on buffer until everything has been uncompressed */ + do { + DBGPRINTF("imptcp: in inflate() loop, avail_in %d, total_in %ld\n", pThis->zstrm.avail_in, pThis->zstrm.total_in); + pThis->zstrm.avail_out = sizeof(zipBuf); + pThis->zstrm.next_out = zipBuf; + zRet = inflate(&pThis->zstrm, Z_NO_FLUSH); /* no bad return value */ + DBGPRINTF("after inflate, ret %d, avail_out %d\n", zRet, pThis->zstrm.avail_out); + outavail = sizeof(zipBuf) - pThis->zstrm.avail_out; + if(outavail != 0) { + outtotal += outavail; + pThis->pLstn->rcvdDecompressed += outavail; + CHKiRet(DataRcvdUncompressed(pThis, (char*)zipBuf, outavail, ttGenTime)); + } + } while (pThis->zstrm.avail_out == 0); + + dbgprintf("end of DataRcvCompress, sizes: in %lld, out %llu\n", (long long) len, outtotal); +finalize_it: + RETiRet; +} + +static rsRetVal +DataRcvd(ptcpsess_t *pThis, char *pData, size_t iLen) +{ + DEFiRet; + pThis->pLstn->rcvdBytes += iLen; + if(pThis->compressionMode >= COMPRESS_STREAM_ALWAYS) + iRet = DataRcvdCompressed(pThis, pData, iLen); + else + iRet = DataRcvdUncompressed(pThis, pData, iLen, 0); + RETiRet; +} + /****************************************** --END-- TCP SUPPORT FUNCTIONS ***********************************/ @@ -936,6 +1015,14 @@ addLstn(ptcpsrv_t *pSrv, int sock, int isIPv6) STATSCOUNTER_INIT(pLstn->ctrSubmit, pLstn->mutCtrSubmit); CHKiRet(statsobj.AddCounter(pLstn->stats, UCHAR_CONSTANT("submitted"), ctrType_IntCtr, &(pLstn->ctrSubmit))); + /* the following counters are not protected by mutexes; we accept + * that they may not be 100% correct */ + pLstn->rcvdBytes = 0, + pLstn->rcvdDecompressed = 0; + CHKiRet(statsobj.AddCounter(pLstn->stats, UCHAR_CONSTANT("bytes.received"), + ctrType_IntCtr, &(pLstn->rcvdBytes))); + CHKiRet(statsobj.AddCounter(pLstn->stats, UCHAR_CONSTANT("bytes.decompressed"), + ctrType_IntCtr, &(pLstn->rcvdDecompressed))); CHKiRet(statsobj.ConstructFinalize(pLstn->stats)); /* add to start of server's listener list */ @@ -948,6 +1035,7 @@ addLstn(ptcpsrv_t *pSrv, int sock, int isIPv6) iRet = addEPollSock(epolld_lstn, pLstn, sock, &pLstn->epd); finalize_it: +dbgprintf("DDDD: addLstn return %d\n", iRet); RETiRet; } @@ -971,6 +1059,7 @@ addSess(ptcplstn_t *pLstn, int sock, prop_t *peerName, prop_t *peerIP) pSess->bAtStrtOfFram = 1; pSess->peerName = peerName; pSess->peerIP = peerIP; + pSess->compressionMode = pLstn->pSrv->compressionMode; /* add to start of server's listener list */ pSess->prev = NULL; @@ -988,6 +1077,44 @@ finalize_it: } +/* finish zlib buffer, to be called before closing the session. + */ +static rsRetVal +doZipFinish(ptcpsess_t *pSess) +{ + int zRet; /* zlib return state */ + DEFiRet; + unsigned outavail; + uchar zipBuf[32*1024]; // TODO: use "global" one from pSess + + if(!pSess->bzInitDone) + goto done; + + pSess->zstrm.avail_in = 0; + /* run inflate() on buffer until everything has been compressed */ + do { + DBGPRINTF("doZipFinish: in inflate() loop, avail_in %d, total_in %ld\n", pSess->zstrm.avail_in, pSess->zstrm.total_in); + pSess->zstrm.avail_out = sizeof(zipBuf); + pSess->zstrm.next_out = zipBuf; + zRet = inflate(&pSess->zstrm, Z_FINISH); /* no bad return value */ + DBGPRINTF("after inflate, ret %d, avail_out %d\n", zRet, pSess->zstrm.avail_out); + outavail = sizeof(zipBuf) - pSess->zstrm.avail_out; + if(outavail != 0) { + pSess->pLstn->rcvdDecompressed += outavail; + CHKiRet(DataRcvdUncompressed(pSess, (char*)zipBuf, outavail, 0)); // TODO: query time! + } + } while (pSess->zstrm.avail_out == 0); + +finalize_it: + zRet = inflateEnd(&pSess->zstrm); + if(zRet != Z_OK) { + DBGPRINTF("imptcp: error %d returned from zlib/inflateEnd()\n", zRet); + } + + pSess->bzInitDone = 0; +done: RETiRet; +} + /* close/remove a session * NOTE: we must first remove the fd from the epoll set and then close it -- else we * get an error "bad file descriptor" from epoll. @@ -998,6 +1125,9 @@ closeSess(ptcpsess_t *pSess) int sock; DEFiRet; + if(pSess->compressionMode >= COMPRESS_STREAM_ALWAYS) + doZipFinish(pSess); + sock = pSess->sock; CHKiRet(removeEPollSock(sock, pSess->epd)); close(sock); @@ -1048,6 +1178,7 @@ createInstance(instanceConf_t **pinst) inst->pBindRuleset = NULL; inst->ratelimitBurst = 10000; /* arbitrary high limit */ inst->ratelimitInterval = 0; /* off */ + inst->compressionMode = COMPRESS_SINGLE_MSG; /* node created, let's add to config */ if(loadModConf->tail == NULL) { @@ -1127,6 +1258,7 @@ addListner(modConfData_t __attribute__((unused)) *modConf, instanceConf_t *inst) pSrv->iKeepAliveProbes = inst->iKeepAliveProbes; pSrv->iKeepAliveTime = inst->iKeepAliveTime; pSrv->bEmitMsgOnClose = inst->bEmitMsgOnClose; + pSrv->compressionMode = inst->compressionMode; CHKiRet(ratelimitNew(&pSrv->ratelimiter, "imtcp", (char*)inst->pszBindPort)); ratelimitSetLinuxLike(pSrv->ratelimiter, inst->ratelimitInterval, inst->ratelimitBurst); ratelimitSetThreadSafe(pSrv->ratelimiter); @@ -1269,6 +1401,10 @@ sessActivity(ptcpsess_t *pSess) { int lenRcv; int lenBuf; + uchar *peerName; + int lenPeer; + int remsock = 0; /* init just to keep compiler happy... :-( */ + sbool bEmitOnClose = 0; char rcvBuf[128*1024]; DEFiRet; @@ -1285,13 +1421,15 @@ sessActivity(ptcpsess_t *pSess) } else if (lenRcv == 0) { /* session was closed, do clean-up */ if(pSess->pLstn->pSrv->bEmitMsgOnClose) { - uchar *peerName; - int lenPeer; - prop.GetString(pSess->peerName, &peerName, &lenPeer); - errmsg.LogError(0, RS_RET_PEER_CLOSED_CONN, "imptcp session %d closed by remote peer %s.\n", - pSess->sock, peerName); + prop.GetString(pSess->peerName, &peerName, &lenPeer), + remsock = pSess->sock; + bEmitOnClose = 1; + } + CHKiRet(closeSess(pSess)); /* close may emit more messages in strmzip mode! */ + if(bEmitOnClose) { + errmsg.LogError(0, RS_RET_PEER_CLOSED_CONN, "imptcp session %d closed by " + "remote peer %s.\n", remsock, peerName); } - CHKiRet(closeSess(pSess)); break; } else { if(errno == EAGAIN || errno == EWOULDBLOCK) @@ -1415,6 +1553,7 @@ wrkr(void *myself) BEGINnewInpInst struct cnfparamvals *pvals; instanceConf_t *inst; + char *cstr; int i; CODESTARTnewInpInst DBGPRINTF("newInpInst (imptcp)\n"); @@ -1446,6 +1585,19 @@ CODESTARTnewInpInst inst->pszBindRuleset = (uchar*)es_str2cstr(pvals[i].val.d.estr, NULL); } else if(!strcmp(inppblk.descr[i].name, "supportoctetcountedframing")) { inst->bSuppOctetFram = (int) pvals[i].val.d.n; + } else if(!strcmp(inppblk.descr[i].name, "compression.mode")) { + cstr = es_str2cstr(pvals[i].val.d.estr, NULL); + if(!strcasecmp(cstr, "stream:always")) { + inst->compressionMode = COMPRESS_STREAM_ALWAYS; + } else if(!strcasecmp(cstr, "none")) { + inst->compressionMode = COMPRESS_NEVER; + } 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 if(!strcmp(inppblk.descr[i].name, "keepalive")) { inst->bKeepAlive = (int) pvals[i].val.d.n; } else if(!strcmp(inppblk.descr[i].name, "keepalive.probes")) { @@ -1653,6 +1805,7 @@ shutdownSrv(ptcpsrv_t *pSrv) ptcplstn_t *pLstn, *lstnDel; ptcpsess_t *pSess, *sessDel; +dbgprintf("DDDD: enter shutdownSrv\n"); /* listeners */ pLstn = pSrv->pLstn; while(pLstn != NULL) { @@ -1661,7 +1814,9 @@ shutdownSrv(ptcpsrv_t *pSrv) /* now unlink listner */ lstnDel = pLstn; pLstn = pLstn->next; - DBGPRINTF("imptcp shutdown listen socket %d\n", lstnDel->sock); + DBGPRINTF("imptcp shutdown listen socket %d (rcvd %lld bytes, " + "decompressed %lld)\n", lstnDel->sock, lstnDel->rcvdBytes, + lstnDel->rcvdDecompressed); free(lstnDel->epd); free(lstnDel); } diff --git a/tools/omfwd.c b/tools/omfwd.c index 129392d2..179e9614 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. * @@ -39,6 +39,7 @@ #include <errno.h> #include <ctype.h> #include <unistd.h> +#include <stdint.h> #ifdef USE_NETZIP #include <zlib.h> #endif @@ -97,6 +98,13 @@ typedef struct _instanceData { TCPFRAMINGMODE tcp_framing; int bResendLastOnRecon; /* should the last message be re-sent on a successful reconnect? */ tcpclt_t *pTCPClt; /* our tcpclt object */ +# 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; + 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; @@ -132,6 +140,7 @@ static struct cnfparamdescr actpdescr[] = { { "protocol", eCmdHdlrGetWord, 0 }, { "tcp_framing", eCmdHdlrGetWord, 0 }, { "ziplevel", eCmdHdlrInt, 0 }, + { "compression.mode", eCmdHdlrGetWord, 0 }, { "rebindinterval", eCmdHdlrInt, 0 }, { "streamdriver", eCmdHdlrGetWord, 0 }, { "streamdrivermode", eCmdHdlrInt, 0 }, @@ -169,6 +178,7 @@ ENDinitConfVars static rsRetVal doTryResume(instanceData *pData); +static rsRetVal doZipFinish(instanceData *pData); /* this function gets the default template. It coordinates action between * old-style and new-style configuration parts. @@ -240,6 +250,7 @@ static inline void DestructTCPInstanceData(instanceData *pData) { assert(pData != NULL); + doZipFinish(pData); if(pData->pNetstrm != NULL) netstrm.Destruct(&pData->pNetstrm); if(pData->pNS != NULL) @@ -423,14 +434,8 @@ 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(instanceData *pData, uchar *buf, unsigned len) { DEFiRet; unsigned alreadySent; @@ -438,6 +443,7 @@ TCPSendBuf(instanceData *pData, uchar *buf, unsigned len) alreadySent = 0; CHKiRet(netstrm.CheckConnection(pData->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)); @@ -455,6 +461,99 @@ finalize_it: RETiRet; } +static rsRetVal +TCPSendBufCompressed(instanceData *pData, uchar *buf, unsigned len) +{ + int zRet; /* zlib return state */ + unsigned outavail; + uchar zipBuf[32*1024]; + DEFiRet; + + if(!pData->bzInitDone) { + /* allocate deflate state */ + pData->zstrm.zalloc = Z_NULL; + pData->zstrm.zfree = Z_NULL; + pData->zstrm.opaque = Z_NULL; + /* see note in file header for the params we use with deflateInit2() */ + zRet = deflateInit(&pData->zstrm, 9); + if(zRet != Z_OK) { + DBGPRINTF("error %d returned from zlib/deflateInit()\n", zRet); + ABORT_FINALIZE(RS_RET_ZLIB_ERR); + } + pData->bzInitDone = RSTRUE; + } + + /* now doing the compression */ + pData->zstrm.next_in = (Bytef*) buf; + pData->zstrm.avail_in = len; + /* run deflate() on buffer until everything has been compressed */ + do { + DBGPRINTF("omfwd: in deflate() loop, avail_in %d, total_in %ld\n", pData->zstrm.avail_in, pData->zstrm.total_in); + pData->zstrm.avail_out = sizeof(zipBuf); + pData->zstrm.next_out = zipBuf; + zRet = deflate(&pData->zstrm, Z_NO_FLUSH); /* no bad return value */ + DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pData->zstrm.avail_out); + outavail = sizeof(zipBuf) - pData->zstrm.avail_out; + if(outavail != 0) { + CHKiRet(TCPSendBufUncompressed(pData, zipBuf, outavail)); + } + } while (pData->zstrm.avail_out == 0); + +finalize_it: + RETiRet; +} + +static rsRetVal +TCPSendBuf(instanceData *pData, uchar *buf, unsigned len) +{ + DEFiRet; + if(pData->compressionMode >= COMPRESS_STREAM_ALWAYS) + iRet = TCPSendBufCompressed(pData, buf, len); + else + iRet = TCPSendBufUncompressed(pData, buf, len); + RETiRet; +} + +/* finish zlib buffer, to be called before closing the ZIP file (if + * running in stream mode). + */ +static rsRetVal +doZipFinish(instanceData *pData) +{ + int zRet; /* zlib return state */ + DEFiRet; + unsigned outavail; + uchar zipBuf[32*1024]; + + if(!pData->bzInitDone) + goto done; + +// TODO: can we get this into a single common function? +dbgprintf("DDDD: in doZipFinish()\n"); + pData->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", pData->zstrm.avail_in, pData->zstrm.total_in); + pData->zstrm.avail_out = sizeof(zipBuf); + pData->zstrm.next_out = zipBuf; + zRet = deflate(&pData->zstrm, Z_FINISH); /* no bad return value */ + DBGPRINTF("after deflate, ret %d, avail_out %d\n", zRet, pData->zstrm.avail_out); + outavail = sizeof(zipBuf) - pData->zstrm.avail_out; + if(outavail != 0) { + CHKiRet(TCPSendBufUncompressed(pData, zipBuf, outavail)); + } + } while (pData->zstrm.avail_out == 0); + +finalize_it: + zRet = deflateEnd(&pData->zstrm); + if(zRet != Z_OK) { + DBGPRINTF("error %d returned from zlib/deflateEnd()\n", zRet); + } + + pData->bzInitDone = 0; +done: RETiRet; +} + /* Add frame to send buffer (or send, if requried) */ @@ -636,11 +735,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'; @@ -756,14 +854,17 @@ setInstParamDefaults(instanceData *pData) pData->iRebindInterval = 0; pData->bResendLastOnRecon = 0; pData->pPermPeers = NULL; - pData->compressionLevel = 0; + pData->compressionLevel = 9; + pData->compressionMode = COMPRESS_NEVER; } BEGINnewActInst struct cnfparamvals *pvals; uchar *tplToUse; + char *cstr; int i; rsRetVal localRet; + int complevel = -1; CODESTARTnewActInst DBGPRINTF("newActInst (omfwd)\n"); @@ -860,9 +961,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.", @@ -876,11 +978,37 @@ CODESTARTnewActInst 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.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); @@ -948,6 +1076,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 +1154,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)! */ |