Merge branch 'master' into v5-devel

Conflicts:
	ChangeLog
	runtime/rsyslog.h

21 files changed, 1641 insertions, 27 deletions

diff --git a/ChangeLog b/ChangeLog
index 4b437a7c..1e954a2a 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -32,9 +32,16 @@ increase.
 - increased ompgsql performance by adapting to new transactional
   output module interface
 ---------------------------------------------------------------------------
-Version 4.3.?  [DEVEL] (rgerhards), 2009-??-??
+Version 4.3.2  [DEVEL] (rgerhards), 2009-??-??
+- added a generic network stream server (in addition to rather specific
+  syslog tcp server)
+- added ability for the UDP output action to rebind its send socket after
+  sending n messages. New config directive $ActionSendUDPRebindInterval
+  added for the purpose. By default, rebinding is disabled. This is 
+  considered useful for load balancers.
 - bugfix: imdiag/imtcp had a race condition
 - improved testbench (now much better code design and reuse)
+- added config switch --enable-testbench=no to turn off testbench
 ---------------------------------------------------------------------------
 Version 4.3.1  [DEVEL] (rgerhards), 2009-05-25
 - added capability to run multiple tcp listeners (on different ports)
diff --git a/Makefile.am b/Makefile.am
index e991f323..8a130655 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -84,6 +84,10 @@ if ENABLE_SNMP
 SUBDIRS += plugins/omsnmp
 endif
 
+if ENABLE_CUST1
+SUBDIRS += plugins/cust1
+endif
+
 if ENABLE_IMTEMPLATE
 SUBDIRS += plugins/imtemplate
 endif
diff --git a/configure.ac b/configure.ac
index ffa1f230..06ceec3c 100644
--- a/configure.ac
+++ b/configure.ac
@@ -678,6 +678,20 @@ AC_SUBST(LIBLOGGING_CFLAGS)
 AC_SUBST(LIBLOGGING_LIBS)
 
 
+# enable/disable the testbench (e.g. because some important parts
+# are missing)
+AC_ARG_ENABLE(testbench,
+        [AS_HELP_STRING([--enable-testbench],[file input module enabled @<:@default=yes@:>@])],
+        [case "${enableval}" in
+         yes) enable_testbench="yes" ;;
+          no) enable_testbench="no" ;;
+           *) AC_MSG_ERROR(bad value ${enableval} for --enable-testbench) ;;
+         esac],
+        [enable_testbench=yes]
+)
+AM_CONDITIONAL(ENABLE_TESTBENCH, test x$enable_testbench = xyes)
+
+
 # settings for the file input module
 AC_ARG_ENABLE(imfile,
         [AS_HELP_STRING([--enable-imfile],[file input module enabled @<:@default=no@:>@])],
@@ -716,6 +730,20 @@ AC_ARG_ENABLE(omstdout,
 )
 AM_CONDITIONAL(ENABLE_OMSTDOUT, test x$enable_omstdout = xyes)
 
+# This provides a vehicle to integrate custom modules, that are not
+# part of rsyslog, into the build process. It is named cust1, so that
+# additional such modules can easily be added.
+AC_ARG_ENABLE(cust1,
+        [AS_HELP_STRING([--enable-cust1],[Compiles stdout module @<:@default=no@:>@])],
+        [case "${enableval}" in
+         yes) enable_cust1="yes" ;;
+          no) enable_cust1="no" ;;
+           *) AC_MSG_ERROR(bad value ${enableval} for --enable-cust1) ;;
+         esac],
+        [enable_cust1=no]
+)
+AM_CONDITIONAL(ENABLE_CUST1, test x$enable_cust1 = xyes)
+
 
 # settings for the template input module; copy and modify this code
 # if you intend to add your own module. Be sure to replace imtemplate
@@ -789,6 +817,7 @@ AC_CONFIG_FILES([Makefile \
 		plugins/ommail/Makefile \
 		plugins/omsnmp/Makefile \
 		plugins/omoracle/Makefile \
+		plugins/cust1/Makefile \
 		tests/Makefile])
 AC_OUTPUT
 
@@ -802,6 +831,7 @@ echo "    Regular expressions support enabled:      $enable_regexp"
 echo "    Zlib compression support enabled:         $enable_zlib"
 echo "    rsyslog runtime will be built:            $enable_rsyslogrt"
 echo "    rsyslogd will be built:                   $enable_rsyslogd"
+echo "    custom module 1 will be built:            $enable_cust1"
 echo
 echo "---{ input plugins }---"
 echo "    Klog functionality enabled:               $enable_klog ($os_type)"
@@ -828,6 +858,7 @@ echo "    RELP support enabled:                     $enable_relp"
 echo "    SNMP support enabled:                     $enable_snmp"
 echo
 echo "---{ debugging support }---"
+echo "    Testbench enabled:                        $enable_testbench"
 echo "    Debug mode enabled:                       $enable_debug"
 echo "    Runtime Instrumentation enabled:          $enable_rtinst"
 echo "    Diagnostic tools enabled:                 $enable_diagtools"
diff --git a/doc/rsyslog_conf_global.html b/doc/rsyslog_conf_global.html
index a3cb8ba5..c7574e5d 100644
--- a/doc/rsyslog_conf_global.html
+++ b/doc/rsyslog_conf_global.html
@@ -97,6 +97,9 @@ default 60000 (1 minute)]</li>
 (driver-specific)</li><li>$ActionSendStreamDriverAuthMode &lt;mode&gt;,&nbsp; authentication mode to use with the stream driver
 (driver-specific)</li><li>$ActionSendStreamDriverPermittedPeer &lt;ID&gt;,&nbsp; accepted fingerprint (SHA1) or name of remote peer
 (driver-specific) -<span style="font-weight: bold;"> directive may go away</span>!</li>
+<li><b>$ActionSendUDPRebindInterval</b> nbr</a>- [available since 4.3.2] - instructs the UDP send
+action to rebind the send socket every nbr of messages sent. Zero, the default, means
+that no rebind is done. This directive is useful for use with load-balancers.</li>
 <li><a href="rsconf1_allowedsender.html">$AllowedSender</a></li>
 <li><a href="rsconf1_controlcharacterescapeprefix.html">$ControlCharacterEscapePrefix</a></li>
 <li><a href="rsconf1_debugprintcfsyslinehandlerlist.html">$DebugPrintCFSyslineHandlerList</a></li>
diff --git a/plugins/cust1/Makefile.am b/plugins/cust1/Makefile.am
new file mode 100644
index 00000000..d2e075f9
--- /dev/null
+++ b/plugins/cust1/Makefile.am
@@ -0,0 +1,6 @@
+pkglib_LTLIBRARIES = cust1.la
+
+cust1_la_SOURCES = cust1.c
+cust1_la_CPPFLAGS = -I$(top_srcdir) $(PTHREADS_CFLAGS) $(RSRT_CFLAGS)
+cust1_la_LDFLAGS = -module -avoid-version
+cust1_la_LIBADD = 
diff --git a/runtime/Makefile.am b/runtime/Makefile.am
index eaae9705..838c15bd 100644
--- a/runtime/Makefile.am
+++ b/runtime/Makefile.am
@@ -122,6 +122,13 @@ lmnetstrms_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS)
 lmnetstrms_la_LDFLAGS = -module -avoid-version
 lmnetstrms_la_LIBADD =
 
+# generic stream server framework
+pkglib_LTLIBRARIES += lmstrmsrv.la
+lmstrmsrv_la_SOURCES = strmsrv.c strmsrv.h strms_sess.c strms_sess.h
+lmstrmsrv_la_CPPFLAGS = $(PTHREADS_CFLAGS) $(RSRT_CFLAGS)
+lmstrmsrv_la_LDFLAGS = -module -avoid-version
+lmstrmsrv_la_LIBADD =
+
 # netstream drivers
 
 # plain tcp driver - main driver
diff --git a/runtime/conf.c b/runtime/conf.c
index 1691e23a..0d0a8bd3 100644
--- a/runtime/conf.c
+++ b/runtime/conf.c
@@ -484,7 +484,7 @@ finalize_it:
 	}
 
 	if(bHadAnError && (iRet == RS_RET_OK)) { /* a bit dirty, enhance in future releases */
-		iRet = RS_RET_ERR;
+		iRet = RS_RET_NONFATAL_CONFIG_ERR;
 	}
 	RETiRet;
 }
diff --git a/runtime/netstrm.c b/runtime/netstrm.c
index 05bb25c0..3658006f 100644
--- a/runtime/netstrm.c
+++ b/runtime/netstrm.c
@@ -233,6 +233,19 @@ Send(netstrm_t *pThis, uchar *pBuf, ssize_t *pLenBuf)
 	RETiRet;
 }
 
+/* Enable Keep-Alive handling for those drivers that support it.
+ * rgerhards, 2009-06-02
+ */
+static rsRetVal
+EnableKeepAlive(netstrm_t *pThis)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, netstrm);
+	iRet = pThis->Drvr.EnableKeepAlive(pThis->pDrvrData);
+	RETiRet;
+}
+
+
 
 /* check connection - slim wrapper for NSD driver function */
 static void
@@ -337,6 +350,7 @@ CODESTARTobjQueryInterface(netstrm)
 	pIf->SetDrvrPermPeers = SetDrvrPermPeers;
 	pIf->CheckConnection = CheckConnection;
 	pIf->GetSock = GetSock;
+	pIf->EnableKeepAlive = EnableKeepAlive;
 finalize_it:
 ENDobjQueryInterface(netstrm)
 
diff --git a/runtime/netstrm.h b/runtime/netstrm.h
index b00dd223..f6931104 100644
--- a/runtime/netstrm.h
+++ b/runtime/netstrm.h
@@ -69,9 +69,13 @@ BEGINinterface(netstrm) /* name must also be changed in ENDinterface macro! */
 	 * sockets - not really desirable, but not the end of the world... TODO: should be
 	 * reconsidered when a new ACL system is build. -- rgerhards, 2008-12-01
 	 */
+	/* v4 */
+	rsRetVal (*EnableKeepAlive)(netstrm_t *pThis);
 ENDinterface(netstrm)
-#define netstrmCURR_IF_VERSION 3 /* increment whenever you change the interface structure! */
-/* interface version 3 added GetRemAddr() */
+#define netstrmCURR_IF_VERSION 4 /* increment whenever you change the interface structure! */
+/* interface version 3 added GetRemAddr()
+ * interface version 4 added EnableKeepAlive() -- rgerhards, 2009-06-02
+ * */
 
 /* prototypes */
 PROTOTYPEObj(netstrm);
diff --git a/runtime/nsd.h b/runtime/nsd.h
index f0c9b9b6..8668c934 100644
--- a/runtime/nsd.h
+++ b/runtime/nsd.h
@@ -69,9 +69,13 @@ BEGINinterface(nsd) /* name must also be changed in ENDinterface macro! */
 	 * sockets - not really desirable, but not the end of the world... TODO: should be
 	 * reconsidered when a new ACL system is build. -- rgerhards, 2008-12-01
 	 */
+	/* v5 */
+	rsRetVal (*EnableKeepAlive)(nsd_t *pThis);
 ENDinterface(nsd)
-#define nsdCURR_IF_VERSION 4 /* increment whenever you change the interface structure! */
-/* interface version 4 added GetRemAddr() */
+#define nsdCURR_IF_VERSION 5 /* increment whenever you change the interface structure! */
+/* interface version 4 added GetRemAddr()
+ * interface version 5 added EnableKeepAlive() -- rgerhards, 2009-06-02
+ */
 
 /* interface  for the select call */
 BEGINinterface(nsdsel) /* name must also be changed in ENDinterface macro! */
diff --git a/runtime/nsd_gtls.c b/runtime/nsd_gtls.c
index 5786e191..1a50e2f8 100644
--- a/runtime/nsd_gtls.c
+++ b/runtime/nsd_gtls.c
@@ -1560,6 +1560,16 @@ finalize_it:
 	RETiRet;
 }
 
+/* Enable KEEPALIVE handling on the socket.
+ * rgerhards, 2009-06-02
+ */
+static rsRetVal
+EnableKeepAlive(nsd_t *pNsd)
+{
+	return nsd_ptcp.EnableKeepAlive(pNsd);
+}
+
+
 
 /* open a connection to a remote host (server). With GnuTLS, we always
  * open a plain tcp socket and then, if in TLS mode, do a handshake on it.
@@ -1669,6 +1679,7 @@ CODESTARTobjQueryInterface(nsd_gtls)
 	pIf->GetRemoteHName = GetRemoteHName;
 	pIf->GetRemoteIP = GetRemoteIP;
 	pIf->GetRemAddr = GetRemAddr;
+	pIf->EnableKeepAlive = EnableKeepAlive;
 finalize_it:
 ENDobjQueryInterface(nsd_gtls)
 
diff --git a/runtime/nsd_ptcp.c b/runtime/nsd_ptcp.c
index cc531ca0..54ee0666 100644
--- a/runtime/nsd_ptcp.c
+++ b/runtime/nsd_ptcp.c
@@ -614,6 +614,34 @@ finalize_it:
 }
 
 
+/* Enable KEEPALIVE handling on the socket.
+ * rgerhards, 2009-06-02
+ */
+static rsRetVal
+EnableKeepAlive(nsd_t *pNsd)
+{
+	nsd_ptcp_t *pThis = (nsd_ptcp_t*) pNsd;
+	int ret;
+	int optval;
+	socklen_t optlen;
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, nsd_ptcp);
+
+	optval = 1;
+	optlen = sizeof(optval);
+	ret = setsockopt(pThis->sock, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen);
+	if(ret < 0) {
+		dbgprintf("EnableKeepAlive socket call returns error %d\n", ret);
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+
+	dbgprintf("KEEPALIVE enabled for nsd %p\n", pThis);
+
+finalize_it:
+	RETiRet;
+}
+
+
 /* open a connection to a remote host (server).
  * rgerhards, 2008-03-19
  */
@@ -754,6 +782,7 @@ CODESTARTobjQueryInterface(nsd_ptcp)
 	pIf->GetRemoteHName = GetRemoteHName;
 	pIf->GetRemoteIP = GetRemoteIP;
 	pIf->CheckConnection = CheckConnection;
+	pIf->EnableKeepAlive = EnableKeepAlive;
 finalize_it:
 ENDobjQueryInterface(nsd_ptcp)
 
diff --git a/runtime/rsyslog.h b/runtime/rsyslog.h
index 58346d03..a39ae780 100644
--- a/runtime/rsyslog.h
+++ b/runtime/rsyslog.h
@@ -101,6 +101,8 @@ typedef struct permittedPeers_s permittedPeers_t; /* this should go away in the
 typedef struct permittedPeerWildcard_s permittedPeerWildcard_t; /* this should go away in the long term -- rgerhards, 2008-05-19 */
 typedef struct tcpsrv_s tcpsrv_t;
 typedef struct tcps_sess_s tcps_sess_t;
+typedef struct strmsrv_s strmsrv_t;
+typedef struct strms_sess_s strms_sess_t;
 typedef struct vmstk_s vmstk_t;
 typedef struct batch_obj_s batch_obj_t;
 typedef struct batch_s batch_t;
@@ -109,6 +111,7 @@ typedef rsRetVal (*prsf_t)(struct vmstk_s*, int);	/* pointer to a RainerScript f
 typedef uint64 qDeqID;	/* queue Dequeue order ID. 32 bits is considered dangerously few */
 
 typedef struct tcpLstnPortList_s tcpLstnPortList_t; // TODO: rename?
+typedef struct strmLstnPortList_s strmLstnPortList_t; // TODO: rename?
 
 
 #ifdef __hpux
@@ -289,6 +292,7 @@ enum rsRetVal_				/** return value. All methods return this if not specified oth
 	RS_RET_DEFER_COMMIT = -2121, /**< output plugin status: not yet committed (an OK state!) */
 	RS_RET_PREVIOUS_COMMITTED = -2122, /**< output plugin status: previous record was committed (an OK state!) */
 	RS_RET_ACTION_FAILED = -2122, /**< action failed and is now suspended (consider this permanent for the time being) */
+	RS_RET_NONFATAL_CONFIG_ERR = -2123, /**< non-fatal error during config processing */
 	RS_RET_FILENAME_INVALID = -2140, /**< filename invalid, not found, no access, ... */
 
 	/* RainerScript error messages (range 1000.. 1999) */
diff --git a/runtime/strms_sess.c b/runtime/strms_sess.c
new file mode 100644
index 00000000..0aeebe03
--- /dev/null
+++ b/runtime/strms_sess.c
@@ -0,0 +1,300 @@
+/* strms_sess.c
+ *
+ * This implements a session of the strmsrv object. For general
+ * comments, see header of strmsrv.c.
+ *
+ * Copyright 2007, 2008, 2009 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#include "config.h"
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <errno.h>
+#include <ctype.h>
+
+#include "rsyslog.h"
+#include "dirty.h"
+#include "module-template.h"
+#include "net.h"
+#include "strmsrv.h"
+#include "strms_sess.h"
+#include "obj.h"
+#include "errmsg.h"
+#include "netstrm.h"
+#include "msg.h"
+#include "datetime.h"
+
+
+/* static data */
+DEFobjStaticHelpers
+DEFobjCurrIf(glbl)
+DEFobjCurrIf(errmsg)
+DEFobjCurrIf(netstrm)
+DEFobjCurrIf(datetime)
+
+static int iMaxLine; /* maximum size of a single message */
+
+/* forward definitions */
+static rsRetVal Close(strms_sess_t *pThis);
+
+
+/* Standard-Constructor */
+BEGINobjConstruct(strms_sess) /* be sure to specify the object type also in END macro! */
+ENDobjConstruct(strms_sess)
+
+
+/* ConstructionFinalizer
+ */
+static rsRetVal
+strms_sessConstructFinalize(strms_sess_t *pThis)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	if(pThis->pSrv->OnSessConstructFinalize != NULL) {
+		CHKiRet(pThis->pSrv->OnSessConstructFinalize(&pThis->pUsr));
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* destructor for the strms_sess object */
+BEGINobjDestruct(strms_sess) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDestruct(strms_sess)
+	if(pThis->pStrm != NULL)
+		netstrm.Destruct(&pThis->pStrm);
+
+	if(pThis->pSrv->pOnSessDestruct != NULL) {
+		pThis->pSrv->pOnSessDestruct(&pThis->pUsr);
+	}
+	/* now destruct our own properties */
+	free(pThis->fromHost);
+	free(pThis->fromHostIP);
+ENDobjDestruct(strms_sess)
+
+
+/* debugprint for the strms_sess object */
+BEGINobjDebugPrint(strms_sess) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDebugPrint(strms_sess)
+ENDobjDebugPrint(strms_sess)
+
+
+/* set property functions */
+/* set the hostname. Note that the caller *hands over* the string. That is,
+ * the caller no longer controls it once SetHost() has received it. Most importantly,
+ * the caller must not free it. -- rgerhards, 2008-04-24
+ */
+static rsRetVal
+SetHost(strms_sess_t *pThis, uchar *pszHost)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	free(pThis->fromHost);
+	pThis->fromHost = pszHost;
+	RETiRet;
+}
+
+/* set the remote host's IP. Note that the caller *hands over* the string. That is,
+ * the caller no longer controls it once SetHostIP() has received it. Most importantly,
+ * the caller must not free it. -- rgerhards, 2008-05-16
+ */
+static rsRetVal
+SetHostIP(strms_sess_t *pThis, uchar *pszHostIP)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	free(pThis->fromHostIP);
+	pThis->fromHostIP = pszHostIP;
+	RETiRet;
+}
+
+static rsRetVal
+SetStrm(strms_sess_t *pThis, netstrm_t *pStrm)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	pThis->pStrm = pStrm;
+	RETiRet;
+}
+
+
+/* set our parent, the strmsrv object */
+static rsRetVal
+SetStrmsrv(strms_sess_t *pThis, strmsrv_t *pSrv)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	ISOBJ_TYPE_assert(pSrv, strmsrv);
+	pThis->pSrv = pSrv;
+	RETiRet;
+}
+
+
+/* set our parent listener info*/
+static rsRetVal
+SetLstnInfo(strms_sess_t *pThis, strmLstnPortList_t *pLstnInfo)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	assert(pLstnInfo != NULL);
+	pThis->pLstnInfo = pLstnInfo;
+	RETiRet;
+}
+
+
+static rsRetVal
+SetUsrP(strms_sess_t *pThis, void *pUsr)
+{
+	DEFiRet;
+	pThis->pUsr = pUsr;
+	RETiRet;
+}
+
+
+static void *
+GetUsrP(strms_sess_t *pThis)
+{
+	return pThis->pUsr;
+}
+
+
+/* Closes a STRM session
+ * No attention is paid to the return code
+ * of close, so potential-double closes are not detected.
+ */
+static rsRetVal
+Close(strms_sess_t *pThis)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	netstrm.Destruct(&pThis->pStrm);
+	free(pThis->fromHost);
+	pThis->fromHost = NULL; /* not really needed, but... */
+	free(pThis->fromHostIP);
+	pThis->fromHostIP = NULL; /* not really needed, but... */
+
+	RETiRet;
+}
+
+
+
+/* Processes the data received via a STRM session. If there
+ * is no other way to handle it, data is discarded.
+ * Input parameter data is the data received, iLen is its
+ * len as returned from recv(). iLen must be 1 or more (that
+ * is errors must be handled by caller!). iSTRMSess must be
+ * the index of the STRM session that received the data.
+ * rgerhards 2005-07-04
+ * And another change while generalizing. We now return either
+ * RS_RET_OK, which means the session should be kept open
+ * or anything else, which means it must be closed.
+ * rgerhards, 2008-03-01
+ */
+static rsRetVal
+DataRcvd(strms_sess_t *pThis, char *pData, size_t iLen)
+{
+	DEFiRet;
+	char *pEnd;
+
+	ISOBJ_TYPE_assert(pThis, strms_sess);
+	assert(pData != NULL);
+	assert(iLen > 0);
+
+	 /* We now copy the message to the session buffer. */
+	pEnd = pData + iLen; /* this is one off, which is intensional */
+
+	while(pData < pEnd) {
+		CHKiRet(pThis->pSrv->OnCharRcvd(pThis, (uchar)*pData++));
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* queryInterface function
+ * rgerhards, 2008-02-29
+ */
+BEGINobjQueryInterface(strms_sess)
+CODESTARTobjQueryInterface(strms_sess)
+	if(pIf->ifVersion != strms_sessCURR_IF_VERSION) { /* check for current version, increment on each change */
+		ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED);
+	}
+
+	/* ok, we have the right interface, so let's fill it
+	 * Please note that we may also do some backwards-compatibility
+	 * work here (if we can support an older interface version - that,
+	 * of course, also affects the "if" above).
+	 */
+	pIf->DebugPrint = strms_sessDebugPrint;
+	pIf->Construct = strms_sessConstruct;
+	pIf->ConstructFinalize = strms_sessConstructFinalize;
+	pIf->Destruct = strms_sessDestruct;
+
+	pIf->Close = Close;
+	pIf->DataRcvd = DataRcvd;
+
+	pIf->SetUsrP = SetUsrP;
+	pIf->GetUsrP = GetUsrP;
+	pIf->SetStrmsrv = SetStrmsrv;
+	pIf->SetLstnInfo = SetLstnInfo;
+	pIf->SetHost = SetHost;
+	pIf->SetHostIP = SetHostIP;
+	pIf->SetStrm = SetStrm;
+finalize_it:
+ENDobjQueryInterface(strms_sess)
+
+
+/* exit our class
+ * rgerhards, 2008-03-10
+ */
+BEGINObjClassExit(strms_sess, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MACRO! */
+CODESTARTObjClassExit(strms_sess)
+	/* release objects we no longer need */
+	objRelease(errmsg, CORE_COMPONENT);
+	objRelease(netstrm, LM_NETSTRMS_FILENAME);
+	objRelease(datetime, CORE_COMPONENT);
+ENDObjClassExit(strms_sess)
+
+
+/* Initialize our class. Must be called as the very first method
+ * before anything else is called inside this class.
+ * rgerhards, 2008-02-29
+ */
+BEGINObjClassInit(strms_sess, 1, OBJ_IS_CORE_MODULE) /* class, version - CHANGE class also in END MACRO! */
+	/* request objects we use */
+	CHKiRet(objUse(errmsg, CORE_COMPONENT));
+	CHKiRet(objUse(netstrm, LM_NETSTRMS_FILENAME));
+	CHKiRet(objUse(datetime, CORE_COMPONENT));
+
+	CHKiRet(objUse(glbl, CORE_COMPONENT));
+	iMaxLine = glbl.GetMaxLine(); /* get maximum size we currently support */
+	objRelease(glbl, CORE_COMPONENT);
+
+	/* set our own handlers */
+	OBJSetMethodHandler(objMethod_DEBUGPRINT, strms_sessDebugPrint);
+	OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strms_sessConstructFinalize);
+ENDObjClassInit(strms_sess)
+
+/* vim:set ai:
+ */
diff --git a/runtime/strms_sess.h b/runtime/strms_sess.h
new file mode 100644
index 00000000..6483c0c3
--- /dev/null
+++ b/runtime/strms_sess.h
@@ -0,0 +1,75 @@
+/* Definitions for strms_sess class. This implements a session of the
+ * generic stream server.
+ *
+ * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#ifndef INCLUDED_STRMS_SESS_H
+#define INCLUDED_STRMS_SESS_H
+
+#include "obj.h"
+
+/* a forward-definition, we are somewhat cyclic */
+struct strmsrv_s;
+
+/* the strms_sess object */
+struct strms_sess_s {
+	BEGINobjInstance;	/* Data to implement generic object - MUST be the first data element! */
+	strmsrv_t *pSrv;	/* pointer back to my server (e.g. for callbacks) */
+	strmLstnPortList_t *pLstnInfo;	/* pointer back to listener info */
+	netstrm_t *pStrm;
+//	uchar *pMsg;		/* message (fragment) received */
+	uchar *fromHost;
+	uchar *fromHostIP;
+	void *pUsr;		/* a user-pointer */
+};
+
+
+/* interfaces */
+BEGINinterface(strms_sess) /* name must also be changed in ENDinterface macro! */
+	INTERFACEObjDebugPrint(strms_sess);
+	rsRetVal (*Construct)(strms_sess_t **ppThis);
+	rsRetVal (*ConstructFinalize)(strms_sess_t __attribute__((unused)) *pThis);
+	rsRetVal (*Destruct)(strms_sess_t **ppThis);
+	rsRetVal (*Close)(strms_sess_t *pThis);
+	rsRetVal (*DataRcvd)(strms_sess_t *pThis, char *pData, size_t iLen);
+	/* set methods */
+	rsRetVal (*SetStrmsrv)(strms_sess_t *pThis, struct strmsrv_s *pSrv);
+	rsRetVal (*SetLstnInfo)(strms_sess_t *pThis, strmLstnPortList_t *pLstnInfo);
+	rsRetVal (*SetUsrP)(strms_sess_t*, void*);
+	void*    (*GetUsrP)(strms_sess_t*);
+	rsRetVal (*SetHost)(strms_sess_t *pThis, uchar*);
+	rsRetVal (*SetHostIP)(strms_sess_t *pThis, uchar*);
+	rsRetVal (*SetStrm)(strms_sess_t *pThis, netstrm_t*);
+	rsRetVal (*SetOnMsgReceive)(strms_sess_t *pThis, rsRetVal (*OnMsgReceive)(strms_sess_t*, uchar*, int));
+ENDinterface(strms_sess)
+#define strms_sessCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+/* interface changes
+ * to version v2, rgerhards, 2009-05-22
+ * - Data structures changed
+ * - SetLstnInfo entry point added
+ */
+
+
+/* prototypes */
+PROTOTYPEObj(strms_sess);
+
+
+#endif /* #ifndef INCLUDED_STRMS_SESS_H */
diff --git a/runtime/strmsrv.c b/runtime/strmsrv.c
new file mode 100644
index 00000000..3dc53a97
--- /dev/null
+++ b/runtime/strmsrv.c
@@ -0,0 +1,966 @@
+/* strmsrv.c
+ *
+ * This builds a basic stream server. It handles connection creation but
+ * not any protocol. Instead, it calls a "data received" entry point of the
+ * caller with any data received, in which case the caller must react accordingly.
+ * This module works together with the netstream drivers.
+ *
+ * There are actually two classes within the stream server code: one is
+ * the strmsrv itself, the other one is its sessions. This is a helper
+ * class to strmsrv.
+ *
+ * File begun on 2009-06-01 by RGerhards based on strmsrv.c. Note that strmsrv is
+ * placed under LGPL, which is possible because I carefully evaluated and
+ * eliminated all those parts of strmsrv which were not written by me.
+ *
+ * TODO: I would consider it useful to migrate tcpsrv.c/tcps_sess.c to this stream
+ * class here. The requires a little bit redesign, but should not be too hard. The
+ * core idea, already begun here, is that we still support lots of callbacks, but
+ * provide "canned" implementations for standard cases. That way, most upper-layer
+ * modules can be kept rather simple and without any extra overhead. Note that
+ * to support this, tcps_sess.c would need to extract the message reception state
+ * machine to a separate module which then is called via the DoCharRcvd() interface
+ * of this class here. -- rgerhards, 2009-06-01
+ *
+ * Copyright 2007, 2008, 2009 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+
+#include "config.h"
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+#include <stdarg.h>
+#include <ctype.h>
+#include <netinet/in.h>
+#include <netdb.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#if HAVE_FCNTL_H
+#include <fcntl.h>
+#endif
+#include "rsyslog.h"
+#include "dirty.h"
+#include "cfsysline.h"
+#include "module-template.h"
+#include "net.h"
+#include "srUtils.h"
+#include "conf.h"
+#include "strmsrv.h"
+#include "obj.h"
+#include "glbl.h"
+#include "netstrms.h"
+#include "netstrm.h"
+#include "nssel.h"
+#include "errmsg.h"
+#include "unicode-helper.h"
+
+MODULE_TYPE_LIB
+
+/* defines */
+#define STRMSESS_MAX_DEFAULT 200 /* default for nbr of strm sessions if no number is given */
+#define STRMLSTN_MAX_DEFAULT 20 /* default for nbr of listeners */
+
+/* static data */
+DEFobjStaticHelpers
+DEFobjCurrIf(conf)
+DEFobjCurrIf(glbl)
+DEFobjCurrIf(strms_sess)
+DEFobjCurrIf(errmsg)
+DEFobjCurrIf(net)
+DEFobjCurrIf(netstrms)
+DEFobjCurrIf(netstrm)
+DEFobjCurrIf(nssel)
+
+/* forward definitions */
+static rsRetVal create_strm_socket(strmsrv_t *pThis);
+
+/* standard callbacks, if the caller did not provide us with them (this helps keep us
+ * flexible while at the same time permits very simple upper-layer modules)
+ */
+/* this shall go into a specific ACL module! */
+static int
+isPermittedHost(struct sockaddr __attribute__((unused)) *addr, char __attribute__((unused)) *fromHostFQDN,
+		void __attribute__((unused)) *pUsrSrv, void __attribute__((unused)) *pUsrSess)
+{
+	return 1;
+}
+
+
+static rsRetVal
+doOpenLstnSocks(strmsrv_t *pSrv)
+{
+	ISOBJ_TYPE_assert(pSrv, strmsrv);
+	return create_strm_socket(pSrv);
+}
+
+
+static rsRetVal
+doRcvData(strms_sess_t *pSess, char *buf, size_t lenBuf, ssize_t *piLenRcvd)
+{
+	DEFiRet;
+	assert(pSess != NULL);
+	assert(piLenRcvd != NULL);
+
+	*piLenRcvd = lenBuf;
+	CHKiRet(netstrm.Rcv(pSess->pStrm, (uchar*) buf, piLenRcvd));
+finalize_it:
+	RETiRet;
+}
+
+static rsRetVal
+onRegularClose(strms_sess_t *pSess)
+{
+	DEFiRet;
+	assert(pSess != NULL);
+
+	/* process any incomplete frames left over */
+	//strms_sess.PrepareClose(pSess);
+	/* Session closed */
+	strms_sess.Close(pSess);
+	RETiRet;
+}
+
+
+static rsRetVal
+onErrClose(strms_sess_t *pSess)
+{
+	DEFiRet;
+	assert(pSess != NULL);
+
+	strms_sess.Close(pSess);
+	RETiRet;
+}
+
+/* ------------------------------ end callbacks ------------------------------ */
+
+/* add new listener port to listener port list
+ * rgerhards, 2009-05-21
+ */
+static inline rsRetVal
+addNewLstnPort(strmsrv_t *pThis, uchar *pszPort)
+{
+	strmLstnPortList_t *pEntry;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	/* create entry */
+	CHKmalloc(pEntry = malloc(sizeof(strmLstnPortList_t)));
+	pEntry->pszPort = pszPort;
+	pEntry->pSrv = pThis;
+	CHKmalloc(pEntry->pszInputName = ustrdup(pThis->pszInputName));
+
+	/* and add to list */
+	pEntry->pNext = pThis->pLstnPorts;
+	pThis->pLstnPorts = pEntry;
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* configure STRM listener settings.
+ * Note: pszPort is handed over to us - the caller MUST NOT free it!
+ * rgerhards, 2008-03-20
+ */
+static rsRetVal
+configureSTRMListen(strmsrv_t *pThis, uchar *pszPort)
+{
+	int i;
+	uchar *pPort = pszPort;
+	DEFiRet;
+
+	assert(pszPort != NULL);
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	/* extract port */
+	i = 0;
+	while(isdigit((int) *pPort)) {
+		i = i * 10 + *pPort++ - '0';
+	}
+
+	if(i >= 0 && i <= 65535) {
+		CHKiRet(addNewLstnPort(pThis, pszPort));
+	} else {
+		errmsg.LogError(0, NO_ERRCODE, "Invalid STRM listen port %s - ignored.\n", pszPort);
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* Initialize the session table
+ * returns 0 if OK, somewhat else otherwise
+ */
+static rsRetVal
+STRMSessTblInit(strmsrv_t *pThis)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	assert(pThis->pSessions == NULL);
+
+	dbgprintf("Allocating buffer for %d STRM sessions.\n", pThis->iSessMax);
+	if((pThis->pSessions = (strms_sess_t **) calloc(pThis->iSessMax, sizeof(strms_sess_t *))) == NULL) {
+		dbgprintf("Error: STRMSessInit() could not alloc memory for STRM session table.\n");
+		ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* find a free spot in the session table. If the table
+ * is full, -1 is returned, else the index of the free
+ * entry (0 or higher).
+ */
+static int
+STRMSessTblFindFreeSpot(strmsrv_t *pThis)
+{
+	register int i;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	for(i = 0 ; i < pThis->iSessMax ; ++i) {
+		if(pThis->pSessions[i] == NULL)
+			break;
+	}
+
+	return((i < pThis->iSessMax) ? i : -1);
+}
+
+
+/* Get the next session index. Free session tables entries are
+ * skipped. This function is provided the index of the last
+ * session entry, or -1 if no previous entry was obtained. It
+ * returns the index of the next session or -1, if there is no
+ * further entry in the table. Please note that the initial call
+ * might as well return -1, if there is no session at all in the
+ * session table.
+ */
+static int
+STRMSessGetNxtSess(strmsrv_t *pThis, int iCurr)
+{
+	register int i;
+
+	BEGINfunc
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	assert(pThis->pSessions != NULL);
+	for(i = iCurr + 1 ; i < pThis->iSessMax ; ++i) {
+		if(pThis->pSessions[i] != NULL)
+			break;
+	}
+
+	ENDfunc
+	return((i < pThis->iSessMax) ? i : -1);
+}
+
+
+/* De-Initialize STRM listner sockets.
+ * This function deinitializes everything, including freeing the
+ * session table. No STRM listen receive operations are permitted
+ * unless the subsystem is reinitialized.
+ * rgerhards, 2007-06-21
+ */
+static void deinit_strm_listener(strmsrv_t *pThis)
+{
+	int i;
+	strmLstnPortList_t *pEntry;
+	strmLstnPortList_t *pDel;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	if(pThis->pSessions != NULL) {
+		/* close all STRM connections! */
+		i = STRMSessGetNxtSess(pThis, -1);
+		while(i != -1) {
+			strms_sess.Destruct(&pThis->pSessions[i]);
+			/* now get next... */
+			i = STRMSessGetNxtSess(pThis, i);
+		}
+		
+		/* we are done with the session table - so get rid of it...  */
+		free(pThis->pSessions);
+		pThis->pSessions = NULL; /* just to make sure... */
+	}
+
+	/* free list of strm listen ports */
+	pEntry = pThis->pLstnPorts;
+	while(pEntry != NULL) {
+		free(pEntry->pszPort);
+		free(pEntry->pszInputName);
+		pDel = pEntry;
+		pEntry = pEntry->pNext;
+		free(pDel);
+	}
+
+	/* finally close our listen streams */
+	for(i = 0 ; i < pThis->iLstnMax ; ++i) {
+		netstrm.Destruct(pThis->ppLstn + i);
+	}
+}
+
+
+/* add a listen socket to our listen socket array. This is a callback
+ * invoked from the netstrm class. -- rgerhards, 2008-04-23
+ */
+static rsRetVal
+addStrmLstn(void *pUsr, netstrm_t *pLstn)
+{
+	strmLstnPortList_t *pPortList = (strmLstnPortList_t *) pUsr;
+	strmsrv_t *pThis = pPortList->pSrv;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	ISOBJ_TYPE_assert(pLstn, netstrm);
+
+	if(pThis->iLstnMax >= STRMLSTN_MAX_DEFAULT)
+		ABORT_FINALIZE(RS_RET_MAX_LSTN_REACHED);
+
+	pThis->ppLstn[pThis->iLstnMax] = pLstn;
+	pThis->ppLstnPort[pThis->iLstnMax] = pPortList;
+	++pThis->iLstnMax;
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* Initialize STRM listener socket for a single port
+ * rgerhards, 2009-05-21
+ */
+static inline rsRetVal
+initSTRMListener(strmsrv_t *pThis, strmLstnPortList_t *pPortEntry)
+{
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	assert(pPortEntry != NULL);
+
+	/* TODO: add capability to specify local listen address! */
+	CHKiRet(netstrm.LstnInit(pThis->pNS, (void*)pPortEntry, addStrmLstn, pPortEntry->pszPort, NULL, pThis->iSessMax));
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* Initialize STRM sockets (for listener) and listens on them */
+static rsRetVal
+create_strm_socket(strmsrv_t *pThis)
+{
+	strmLstnPortList_t *pEntry;
+	DEFiRet;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	/* init all configured ports */
+	pEntry = pThis->pLstnPorts;
+	while(pEntry != NULL) {
+		CHKiRet(initSTRMListener(pThis, pEntry));
+		pEntry = pEntry->pNext;
+	}
+
+	/* OK, we had success. Now it is also time to
+	 * initialize our connections
+	 */
+	if(STRMSessTblInit(pThis) != 0) {
+		/* OK, we are in some trouble - we could not initialize the
+		 * session table, so we can not continue. We need to free all
+		 * we have assigned so far, because we can not really use it...
+		 */
+		errmsg.LogError(0, RS_RET_ERR, "Could not initialize STRM session table, suspending STRM message reception.");
+		ABORT_FINALIZE(RS_RET_ERR);
+	}
+
+finalize_it:
+	RETiRet;
+}
+
+
+/* Accept new STRM connection; make entry in session table. If there
+ * is no more space left in the connection table, the new STRM
+ * connection is immediately dropped.
+ * ppSess has a pointer to the newly created session, if it succeeds.
+ * If it does not succeed, no session is created and ppSess is
+ * undefined. If the user has provided an OnSessAccept Callback,
+ * this one is executed immediately after creation of the 
+ * session object, so that it can do its own initialization.
+ * rgerhards, 2008-03-02
+ */
+static rsRetVal
+SessAccept(strmsrv_t *pThis, strmLstnPortList_t *pLstnInfo, strms_sess_t **ppSess, netstrm_t *pStrm)
+{
+	DEFiRet;
+	strms_sess_t *pSess = NULL;
+	netstrm_t *pNewStrm = NULL;
+	int iSess = -1;
+	struct sockaddr_storage *addr;
+	uchar *fromHostFQDN = NULL;
+	uchar *fromHostIP = NULL;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	assert(pLstnInfo != NULL);
+
+	CHKiRet(netstrm.AcceptConnReq(pStrm, &pNewStrm));
+
+	/* Add to session list */
+	iSess = STRMSessTblFindFreeSpot(pThis);
+	if(iSess == -1) {
+		errno = 0;
+		errmsg.LogError(0, RS_RET_MAX_SESS_REACHED, "too many strm sessions - dropping incoming request");
+		ABORT_FINALIZE(RS_RET_MAX_SESS_REACHED);
+	}
+
+	if(pThis->bUseKeepAlive) {
+		CHKiRet(netstrm.EnableKeepAlive(pNewStrm));
+	}
+
+	/* we found a free spot and can construct our session object */
+	CHKiRet(strms_sess.Construct(&pSess));
+	CHKiRet(strms_sess.SetStrmsrv(pSess, pThis));
+	CHKiRet(strms_sess.SetLstnInfo(pSess, pLstnInfo));
+
+	/* get the host name */
+	CHKiRet(netstrm.GetRemoteHName(pNewStrm, &fromHostFQDN));
+	CHKiRet(netstrm.GetRemoteIP(pNewStrm, &fromHostIP));
+	CHKiRet(netstrm.GetRemAddr(pNewStrm, &addr));
+	/* TODO: check if we need to strip the domain name here -- rgerhards, 2008-04-24 */
+
+	/* Here we check if a host is permitted to send us messages. If it isn't, we do not further
+	 * process the message but log a warning (if we are configured to do this).
+	 * rgerhards, 2005-09-26
+	 */
+	if(pThis->pIsPermittedHost != NULL
+	   && !pThis->pIsPermittedHost((struct sockaddr*) addr, (char*) fromHostFQDN, pThis->pUsr, pSess->pUsr)) {
+		dbgprintf("%s is not an allowed sender\n", fromHostFQDN);
+		if(glbl.GetOption_DisallowWarning()) {
+			errno = 0;
+			errmsg.LogError(0, RS_RET_HOST_NOT_PERMITTED, "STRM message from disallowed sender %s discarded", fromHostFQDN);
+		}
+		ABORT_FINALIZE(RS_RET_HOST_NOT_PERMITTED);
+	}
+
+	/* OK, we have an allowed sender, so let's continue, what
+	 * means we can finally fill in the session object.
+	 */
+	CHKiRet(strms_sess.SetHost(pSess, fromHostFQDN));
+	fromHostFQDN = NULL; /* we handed this string over */
+	CHKiRet(strms_sess.SetHostIP(pSess, fromHostIP));
+	fromHostIP = NULL; /* we handed this string over */
+	CHKiRet(strms_sess.SetStrm(pSess, pNewStrm));
+	pNewStrm = NULL; /* prevent it from being freed in error handler, now done in strms_sess! */
+	CHKiRet(strms_sess.ConstructFinalize(pSess));
+
+	/* check if we need to call our callback */
+	if(pThis->pOnSessAccept != NULL) {
+		CHKiRet(pThis->pOnSessAccept(pThis, pSess));
+	}
+
+	*ppSess = pSess;
+	pThis->pSessions[iSess] = pSess;
+	pSess = NULL; /* this is now also handed over */
+
+finalize_it:
+	if(iRet != RS_RET_OK) {
+		if(pSess != NULL)
+			strms_sess.Destruct(&pSess);
+		if(pNewStrm != NULL)
+			netstrm.Destruct(&pNewStrm);
+		free(fromHostFQDN);
+		free(fromHostIP);
+	}
+
+	RETiRet;
+}
+
+
+static void
+RunCancelCleanup(void *arg)
+{
+	nssel_t **ppSel = (nssel_t**) arg;
+
+	if(*ppSel != NULL)
+		nssel.Destruct(ppSel);
+}
+
+
+/* This function is called to gather input. */
+#pragma GCC diagnostic ignored "-Wempty-body"
+static rsRetVal
+Run(strmsrv_t *pThis)
+{
+	DEFiRet;
+	int nfds;
+	int i;
+	int iSTRMSess;
+	int bIsReady;
+	strms_sess_t *pNewSess;
+	nssel_t *pSel;
+	ssize_t iRcvd;
+
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	/* this is an endless loop - it is terminated by the framework canelling
+	 * this thread. Thus, we also need to instantiate a cancel cleanup handler
+	 * to prevent us from leaking anything. -- rgerharsd, 20080-04-24
+	 */
+	pthread_cleanup_push(RunCancelCleanup, (void*) &pSel);
+	while(1) {
+		CHKiRet(nssel.Construct(&pSel));
+		// TODO: set driver
+		CHKiRet(nssel.ConstructFinalize(pSel));
+
+		/* Add the STRM listen sockets to the list of read descriptors. */
+		for(i = 0 ; i < pThis->iLstnMax ; ++i) {
+			CHKiRet(nssel.Add(pSel, pThis->ppLstn[i], NSDSEL_RD));
+		}
+
+		/* do the sessions */
+		iSTRMSess = STRMSessGetNxtSess(pThis, -1);
+		while(iSTRMSess != -1) {
+			/* TODO: access to pNsd is NOT really CLEAN, use method... */
+			CHKiRet(nssel.Add(pSel, pThis->pSessions[iSTRMSess]->pStrm, NSDSEL_RD));
+			/* now get next... */
+			iSTRMSess = STRMSessGetNxtSess(pThis, iSTRMSess);
+		}
+
+		/* wait for io to become ready */
+		CHKiRet(nssel.Wait(pSel, &nfds));
+
+		for(i = 0 ; i < pThis->iLstnMax ; ++i) {
+			CHKiRet(nssel.IsReady(pSel, pThis->ppLstn[i], NSDSEL_RD, &bIsReady, &nfds));
+			if(bIsReady) {
+				dbgprintf("New connect on NSD %p.\n", pThis->ppLstn[i]);
+				SessAccept(pThis, pThis->ppLstnPort[i], &pNewSess, pThis->ppLstn[i]);
+				--nfds; /* indicate we have processed one */
+			}
+		}
+
+		/* now check the sessions */
+		iSTRMSess = STRMSessGetNxtSess(pThis, -1);
+		while(nfds && iSTRMSess != -1) {
+			CHKiRet(nssel.IsReady(pSel, pThis->pSessions[iSTRMSess]->pStrm, NSDSEL_RD, &bIsReady, &nfds));
+			if(bIsReady) {
+				char buf[8*1024]; /* reception buffer - may hold a partial or multiple messages */
+				dbgprintf("netstream %p with new data\n", pThis->pSessions[iSTRMSess]->pStrm);
+
+				/* Receive message */
+				iRet = pThis->pRcvData(pThis->pSessions[iSTRMSess], buf, sizeof(buf), &iRcvd);
+				switch(iRet) {
+				case RS_RET_CLOSED:
+					pThis->pOnRegularClose(pThis->pSessions[iSTRMSess]);
+					strms_sess.Destruct(&pThis->pSessions[iSTRMSess]);
+					break;
+				case RS_RET_RETRY:
+					/* we simply ignore retry - this is not an error, but we also have not received anything */
+					break;
+				case RS_RET_OK:
+					/* valid data received, process it! */
+					if(strms_sess.DataRcvd(pThis->pSessions[iSTRMSess], buf, iRcvd) != RS_RET_OK) {
+						/* in this case, something went awfully wrong.
+						 * We are instructed to terminate the session.
+						 */
+						errmsg.LogError(0, NO_ERRCODE, "Tearing down STRM Session %d - see "
+							    "previous messages for reason(s)\n", iSTRMSess);
+						pThis->pOnErrClose(pThis->pSessions[iSTRMSess]);
+						strms_sess.Destruct(&pThis->pSessions[iSTRMSess]);
+					}
+					break;
+				default:
+					errno = 0;
+					errmsg.LogError(0, iRet, "netstream session %p will be closed due to error\n",
+							pThis->pSessions[iSTRMSess]->pStrm);
+					pThis->pOnErrClose(pThis->pSessions[iSTRMSess]);
+					strms_sess.Destruct(&pThis->pSessions[iSTRMSess]);
+					break;
+				}
+				--nfds; /* indicate we have processed one */
+			}
+			iSTRMSess = STRMSessGetNxtSess(pThis, iSTRMSess);
+		}
+		CHKiRet(nssel.Destruct(&pSel));
+finalize_it: /* this is a very special case - this time only we do not exit the function,
+	      * because that would not help us either. So we simply retry it. Let's see
+	      * if that actually is a better idea. Exiting the loop wasn't we always
+	      * crashed, which made sense (the rest of the engine was not prepared for
+	      * that) -- rgerhards, 2008-05-19
+	      */
+		/*EMPTY*/;
+	}
+
+	/* note that this point is usually not reached */
+	pthread_cleanup_pop(0); /* remove cleanup handler */
+
+	RETiRet;
+}
+#pragma GCC diagnostic warning "-Wempty-body"
+
+
+/* Standard-Constructor */
+BEGINobjConstruct(strmsrv) /* be sure to specify the object type also in END macro! */
+	pThis->iSessMax = STRMSESS_MAX_DEFAULT; /* TODO: useful default ;) */
+	/* set default callbacks (used if caller does not overwrite them) */
+	pThis->pIsPermittedHost = isPermittedHost;
+	pThis->OpenLstnSocks = doOpenLstnSocks;
+	pThis->pRcvData = doRcvData;
+	pThis->pOnRegularClose = onRegularClose;
+	pThis->pOnErrClose = onErrClose;
+	/* session specific callbacks */
+	//pThis->OnSessConstructFinalize =
+	//pThis->pOnSessDestruct =
+ENDobjConstruct(strmsrv)
+
+
+/* ConstructionFinalizer */
+static rsRetVal
+strmsrvConstructFinalize(strmsrv_t *pThis)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+
+	/* prepare network stream subsystem */
+	CHKiRet(netstrms.Construct(&pThis->pNS));
+	CHKiRet(netstrms.SetDrvrMode(pThis->pNS, pThis->iDrvrMode));
+	if(pThis->pszDrvrAuthMode != NULL)
+		CHKiRet(netstrms.SetDrvrAuthMode(pThis->pNS, pThis->pszDrvrAuthMode));
+	if(pThis->pPermPeers != NULL)
+		CHKiRet(netstrms.SetDrvrPermPeers(pThis->pNS, pThis->pPermPeers));
+	// TODO: set driver!
+	CHKiRet(netstrms.ConstructFinalize(pThis->pNS));
+
+	/* set up listeners */
+	CHKmalloc(pThis->ppLstn = calloc(STRMLSTN_MAX_DEFAULT, sizeof(netstrm_t*)));
+	CHKmalloc(pThis->ppLstnPort = calloc(STRMLSTN_MAX_DEFAULT, sizeof(strmLstnPortList_t*)));
+	iRet = pThis->OpenLstnSocks(pThis);
+
+finalize_it:
+	if(iRet != RS_RET_OK) {
+		if(pThis->pNS != NULL)
+			netstrms.Destruct(&pThis->pNS);
+	}
+	RETiRet;
+}
+
+
+/* destructor for the strmsrv object */
+BEGINobjDestruct(strmsrv) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDestruct(strmsrv)
+	if(pThis->OnDestruct != NULL)
+		pThis->OnDestruct(pThis->pUsr);
+
+	deinit_strm_listener(pThis);
+
+	if(pThis->pNS != NULL)
+		netstrms.Destruct(&pThis->pNS);
+	free(pThis->pszDrvrAuthMode);
+	free(pThis->ppLstn);
+	free(pThis->ppLstnPort);
+	free(pThis->pszInputName);
+ENDobjDestruct(strmsrv)
+
+
+/* debugprint for the strmsrv object */
+BEGINobjDebugPrint(strmsrv) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDebugPrint(strmsrv)
+ENDobjDebugPrint(strmsrv)
+
+/* set functions */
+static rsRetVal
+SetCBIsPermittedHost(strmsrv_t *pThis, int (*pCB)(struct sockaddr *addr, char *fromHostFQDN, void*, void*))
+{
+	DEFiRet;
+	pThis->pIsPermittedHost = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOnSessAccept(strmsrv_t *pThis, rsRetVal (*pCB)(strmsrv_t*, strms_sess_t*))
+{
+	DEFiRet;
+	pThis->pOnSessAccept = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOnDestruct(strmsrv_t *pThis, rsRetVal (*pCB)(void*))
+{
+	DEFiRet;
+	pThis->OnDestruct = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOnSessConstructFinalize(strmsrv_t *pThis, rsRetVal (*pCB)(void*))
+{
+	DEFiRet;
+	pThis->OnSessConstructFinalize = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOnSessDestruct(strmsrv_t *pThis, rsRetVal (*pCB)(void*))
+{
+	DEFiRet;
+	pThis->pOnSessDestruct = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOnRegularClose(strmsrv_t *pThis, rsRetVal (*pCB)(strms_sess_t*))
+{
+	DEFiRet;
+	pThis->pOnRegularClose = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOnErrClose(strmsrv_t *pThis, rsRetVal (*pCB)(strms_sess_t*))
+{
+	DEFiRet;
+	pThis->pOnErrClose = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetCBOpenLstnSocks(strmsrv_t *pThis, rsRetVal (*pCB)(strmsrv_t*))
+{
+	DEFiRet;
+	pThis->OpenLstnSocks = pCB;
+	RETiRet;
+}
+
+static rsRetVal
+SetUsrP(strmsrv_t *pThis, void *pUsr)
+{
+	DEFiRet;
+	pThis->pUsr = pUsr;
+	RETiRet;
+}
+
+static rsRetVal
+SetKeepAlive(strmsrv_t *pThis, int iVal)
+{
+	DEFiRet;
+	dbgprintf("keep-alive set to %d\n", iVal);
+	pThis->bUseKeepAlive = iVal;
+	RETiRet;
+}
+
+static rsRetVal
+SetOnCharRcvd(strmsrv_t *pThis, rsRetVal (*OnCharRcvd)(strms_sess_t*, uchar))
+{
+	DEFiRet;
+	assert(OnCharRcvd != NULL);
+	pThis->OnCharRcvd = OnCharRcvd;
+	RETiRet;
+}
+
+/* Set the input name to use -- rgerhards, 2008-12-10 */
+static rsRetVal
+SetInputName(strmsrv_t *pThis, uchar *name)
+{
+	uchar *pszName;
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	if(name == NULL)
+		pszName = NULL;
+	else
+		CHKmalloc(pszName = ustrdup(name));
+	free(pThis->pszInputName);
+	pThis->pszInputName = pszName;
+finalize_it:
+	RETiRet;
+}
+
+
+/* here follows a number of methods that shuffle authentication settings down
+ * to the drivers. Drivers not supporting these settings may return an error
+ * state.
+ * -------------------------------------------------------------------------- */   
+
+/* set the driver mode -- rgerhards, 2008-04-30 */
+static rsRetVal
+SetDrvrMode(strmsrv_t *pThis, int iMode)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	pThis->iDrvrMode = iMode;
+	RETiRet;
+}
+
+
+/* set the driver authentication mode -- rgerhards, 2008-05-19 */
+static rsRetVal
+SetDrvrAuthMode(strmsrv_t *pThis, uchar *mode)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	CHKmalloc(pThis->pszDrvrAuthMode = ustrdup(mode));
+finalize_it:
+	RETiRet;
+}
+
+
+/* set the driver's permitted peers -- rgerhards, 2008-05-19 */
+static rsRetVal
+SetDrvrPermPeers(strmsrv_t *pThis, permittedPeers_t *pPermPeers)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	pThis->pPermPeers = pPermPeers;
+	RETiRet;
+}
+
+
+/* End of methods to shuffle autentication settings to the driver.;
+
+ * -------------------------------------------------------------------------- */
+
+
+/* set max number of sessions
+ * this must be called before ConstructFinalize, or it will have no effect!
+ * rgerhards, 2009-04-09
+ */
+static rsRetVal
+SetSessMax(strmsrv_t *pThis, int iMax)
+{
+	DEFiRet;
+	ISOBJ_TYPE_assert(pThis, strmsrv);
+	pThis->iSessMax = iMax;
+	RETiRet;
+}
+
+
+/* queryInterface function
+ * rgerhards, 2008-02-29
+ */
+BEGINobjQueryInterface(strmsrv)
+CODESTARTobjQueryInterface(strmsrv)
+	if(pIf->ifVersion != strmsrvCURR_IF_VERSION) { /* check for current version, increment on each change */
+		ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED);
+	}
+
+	/* ok, we have the right interface, so let's fill it
+	 * Please note that we may also do some backwards-compatibility
+	 * work here (if we can support an older interface version - that,
+	 * of course, also affects the "if" above).
+	 */
+	pIf->DebugPrint = strmsrvDebugPrint;
+	pIf->Construct = strmsrvConstruct;
+	pIf->ConstructFinalize = strmsrvConstructFinalize;
+	pIf->Destruct = strmsrvDestruct;
+
+	pIf->configureSTRMListen = configureSTRMListen;
+	pIf->create_strm_socket = create_strm_socket;
+	pIf->Run = Run;
+
+	pIf->SetKeepAlive = SetKeepAlive;
+	pIf->SetUsrP = SetUsrP;
+	pIf->SetInputName = SetInputName;
+	pIf->SetSessMax = SetSessMax;
+	pIf->SetDrvrMode = SetDrvrMode;
+	pIf->SetDrvrAuthMode = SetDrvrAuthMode;
+	pIf->SetDrvrPermPeers = SetDrvrPermPeers;
+	pIf->SetCBIsPermittedHost = SetCBIsPermittedHost;
+	pIf->SetCBOpenLstnSocks = SetCBOpenLstnSocks;
+	pIf->SetCBOnSessAccept = SetCBOnSessAccept;
+	pIf->SetCBOnSessConstructFinalize = SetCBOnSessConstructFinalize;
+	pIf->SetCBOnSessDestruct = SetCBOnSessDestruct;
+	pIf->SetCBOnDestruct = SetCBOnDestruct;
+	pIf->SetCBOnRegularClose = SetCBOnRegularClose;
+	pIf->SetCBOnErrClose = SetCBOnErrClose;
+	pIf->SetOnCharRcvd = SetOnCharRcvd;
+
+finalize_it:
+ENDobjQueryInterface(strmsrv)
+
+
+/* exit our class
+ * rgerhards, 2008-03-10
+ */
+BEGINObjClassExit(strmsrv, OBJ_IS_LOADABLE_MODULE) /* CHANGE class also in END MACRO! */
+CODESTARTObjClassExit(strmsrv)
+	/* release objects we no longer need */
+	objRelease(strms_sess, DONT_LOAD_LIB);
+	objRelease(conf, CORE_COMPONENT);
+	objRelease(glbl, CORE_COMPONENT);
+	objRelease(errmsg, CORE_COMPONENT);
+	objRelease(netstrms, DONT_LOAD_LIB);
+	objRelease(nssel, DONT_LOAD_LIB);
+	objRelease(netstrm, LM_NETSTRMS_FILENAME);
+	objRelease(net, LM_NET_FILENAME);
+ENDObjClassExit(strmsrv)
+
+
+/* Initialize our class. Must be called as the very first method
+ * before anything else is called inside this class.
+ * rgerhards, 2008-02-29
+ */
+BEGINObjClassInit(strmsrv, 1, OBJ_IS_LOADABLE_MODULE) /* class, version - CHANGE class also in END MACRO! */
+	/* request objects we use */
+	CHKiRet(objUse(errmsg, CORE_COMPONENT));
+	CHKiRet(objUse(net, LM_NET_FILENAME));
+	CHKiRet(objUse(netstrms, LM_NETSTRMS_FILENAME));
+	CHKiRet(objUse(netstrm, DONT_LOAD_LIB));
+	CHKiRet(objUse(nssel, DONT_LOAD_LIB));
+	CHKiRet(objUse(strms_sess, DONT_LOAD_LIB));
+	CHKiRet(objUse(conf, CORE_COMPONENT));
+	CHKiRet(objUse(glbl, CORE_COMPONENT));
+
+	/* set our own handlers */
+	OBJSetMethodHandler(objMethod_DEBUGPRINT, strmsrvDebugPrint);
+	OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strmsrvConstructFinalize);
+ENDObjClassInit(strmsrv)
+
+
+/* --------------- here now comes the plumbing that makes as a library module --------------- */
+
+
+BEGINmodExit
+CODESTARTmodExit
+	/* de-init in reverse order! */
+	strmsrvClassExit();
+	strms_sessClassExit();
+ENDmodExit
+
+
+BEGINqueryEtryPt
+CODESTARTqueryEtryPt
+CODEqueryEtryPt_STD_LIB_QUERIES
+ENDqueryEtryPt
+
+
+BEGINmodInit()
+CODESTARTmodInit
+	*ipIFVersProvided = CURR_MOD_IF_VERSION; /* we only support the current interface specification */
+
+	/* Initialize all classes that are in our module - this includes ourselfs */
+	CHKiRet(strms_sessClassInit(pModInfo));
+	CHKiRet(strmsrvClassInit(pModInfo)); /* must be done after strms_sess, as we use it */
+ENDmodInit
+
+/* vim:set ai:
+ */
diff --git a/runtime/strmsrv.h b/runtime/strmsrv.h
new file mode 100644
index 00000000..86e529c3
--- /dev/null
+++ b/runtime/strmsrv.h
@@ -0,0 +1,112 @@
+/* Definitions for strmsrv class.
+ *
+ * Copyright 2008, 2009 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#ifndef INCLUDED_STRMSRV_H
+#define INCLUDED_STRMSRV_H
+
+#include "obj.h"
+#include "strms_sess.h"
+
+/* list of strm listen ports */
+struct strmLstnPortList_s {
+	uchar *pszPort;			/**< the ports the listener shall listen on */
+	uchar *pszInputName;		/**< value to be used as input name */
+	strmsrv_t *pSrv;			/**< pointer to higher-level server instance */
+	strmLstnPortList_t *pNext;	/**< next port or NULL */
+};
+
+
+/* the strmsrv object */
+struct strmsrv_s {
+	BEGINobjInstance;	/**< Data to implement generic object - MUST be the first data element! */
+	int bUseKeepAlive;	/**< use socket layer KEEPALIVE handling? */
+	netstrms_t *pNS;	/**< pointer to network stream subsystem */
+	int iDrvrMode;		/**< mode of the stream driver to use */
+	uchar *pszDrvrAuthMode;	/**< auth mode of the stream driver to use */
+	uchar *pszInputName;	/**< value to be used as input name */
+	permittedPeers_t *pPermPeers;/**< driver's permitted peers */
+	int iLstnMax;		/**< max nbr of listeners currently supported */
+	netstrm_t **ppLstn;	/**< our netstream listners */
+	strmLstnPortList_t **ppLstnPort; /**< pointer to relevant listen port description */
+	int iSessMax;		/**< max number of sessions supported */
+	strmLstnPortList_t *pLstnPorts;	/**< head pointer for listen ports */
+	int addtlFrameDelim;	/**< additional frame delimiter for plain STRM syslog framing (e.g. to handle NetScreen) */
+	strms_sess_t **pSessions;/**< array of all of our sessions */
+	void *pUsr;		/**< a user-settable pointer (provides extensibility for "derived classes")*/
+	/* callbacks */
+	int      (*pIsPermittedHost)(struct sockaddr *addr, char *fromHostFQDN, void*pUsrSrv, void*pUsrSess);
+	rsRetVal (*pRcvData)(strms_sess_t*, char*, size_t, ssize_t *);
+	rsRetVal (*OpenLstnSocks)(struct strmsrv_s*);
+	rsRetVal (*pOnListenDeinit)(void*);
+	rsRetVal (*OnDestruct)(void*);
+	rsRetVal (*pOnRegularClose)(strms_sess_t *pSess);
+	rsRetVal (*pOnErrClose)(strms_sess_t *pSess);
+	/* session specific callbacks */
+	rsRetVal (*pOnSessAccept)(strmsrv_t *, strms_sess_t*);
+	rsRetVal (*OnSessConstructFinalize)(void*);
+	rsRetVal (*pOnSessDestruct)(void*);
+	rsRetVal (*OnCharRcvd)(strms_sess_t*, uchar);
+};
+
+
+/* interfaces */
+BEGINinterface(strmsrv) /* name must also be changed in ENDinterface macro! */
+	INTERFACEObjDebugPrint(strmsrv);
+	rsRetVal (*Construct)(strmsrv_t **ppThis);
+	rsRetVal (*ConstructFinalize)(strmsrv_t __attribute__((unused)) *pThis);
+	rsRetVal (*Destruct)(strmsrv_t **ppThis);
+	rsRetVal (*configureSTRMListen)(strmsrv_t*, uchar *pszPort);
+	//rsRetVal (*SessAccept)(strmsrv_t *pThis, strmLstnPortList_t*, strms_sess_t **ppSess, netstrm_t *pStrm);
+	rsRetVal (*create_strm_socket)(strmsrv_t *pThis);
+	rsRetVal (*Run)(strmsrv_t *pThis);
+	/* set methods */
+	rsRetVal (*SetAddtlFrameDelim)(strmsrv_t*, int);
+	rsRetVal (*SetInputName)(strmsrv_t*, uchar*);
+	rsRetVal (*SetKeepAlive)(strmsrv_t*, int);
+	rsRetVal (*SetUsrP)(strmsrv_t*, void*);
+	rsRetVal (*SetCBIsPermittedHost)(strmsrv_t*, int (*) (struct sockaddr *addr, char*, void*, void*));
+	rsRetVal (*SetCBOpenLstnSocks)(strmsrv_t *, rsRetVal (*)(strmsrv_t*));
+	rsRetVal (*SetCBOnDestruct)(strmsrv_t*, rsRetVal (*) (void*));
+	rsRetVal (*SetCBOnRegularClose)(strmsrv_t*, rsRetVal (*) (strms_sess_t*));
+	rsRetVal (*SetCBOnErrClose)(strmsrv_t*, rsRetVal (*) (strms_sess_t*));
+	rsRetVal (*SetDrvrMode)(strmsrv_t *pThis, int iMode);
+	rsRetVal (*SetDrvrAuthMode)(strmsrv_t *pThis, uchar *pszMode);
+	rsRetVal (*SetDrvrPermPeers)(strmsrv_t *pThis, permittedPeers_t*);
+	/* session specifics */
+	rsRetVal (*SetCBOnSessAccept)(strmsrv_t*, rsRetVal (*) (strmsrv_t*, strms_sess_t*));
+	rsRetVal (*SetCBOnSessDestruct)(strmsrv_t*, rsRetVal (*) (void*));
+	rsRetVal (*SetCBOnSessConstructFinalize)(strmsrv_t*, rsRetVal (*) (void*));
+	rsRetVal (*SetSessMax)(strmsrv_t *pThis, int iMaxSess);
+	rsRetVal (*SetOnCharRcvd)(strmsrv_t *pThis, rsRetVal (*OnMsgCharRcvd)(strms_sess_t*, uchar));
+ENDinterface(strmsrv)
+#define strmsrvCURR_IF_VERSION 1 /* increment whenever you change the interface structure! */
+/* change for v?:
+ */
+
+
+/* prototypes */
+PROTOTYPEObj(strmsrv);
+
+/* the name of our library binary */
+#define LM_STRMSRV_FILENAME "lmstrmsrv"
+
+#endif /* #ifndef INCLUDED_STRMSRV_H */
diff --git a/tcpsrv.c b/tcpsrv.c
index 02eee88e..3516b2e3 100644
--- a/tcpsrv.c
+++ b/tcpsrv.c
@@ -1,8 +1,11 @@
 /* tcpsrv.c
  *
- * Common code for plain TCP based servers. This is currently being
+ * Common code for plain TCP syslog based servers. This is currently being
  * utilized by imtcp and imgssapi.
  *
+ * NOTE: this is *not* a generic TCP server, but one for syslog servers. For
+ *       generic stream servers, please use ./runtime/strmsrv.c!
+ *
  * There are actually two classes within the tcpserver code: one is
  * the tcpsrv itself, the other one is its sessions. This is a helper
  * class to tcpsrv.
diff --git a/tests/Makefile.am b/tests/Makefile.am
index dfcbb598..04008a1b 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -1,3 +1,4 @@
+if ENABLE_TESTBENCH
 TESTRUNS = rt_init rscript
 check_PROGRAMS = $(TESTRUNS) ourtail nettester tcpflood chkseq
 TESTS = $(TESTRUNS) cfg.sh arrayqueue.sh linkedlistqueue.sh da-mainmsg-q.sh \
@@ -11,6 +12,8 @@ if ENABLE_OMSTDOUT
 TESTS += omod-if-array.sh parsertest.sh inputname.sh fieldtest.sh
 endif
 
+endif # if ENABLE_TESTBENCH
+
 TESTS_ENVIRONMENT = RSYSLOG_MODDIR='$(abs_top_builddir)'/runtime/.libs/
 DISTCLEANFILES=rsyslog.pid '$(abs_top_builddir)'/DiagTalker.class
 test_files = testbench.h runtime-dummy.c
diff --git a/tools/omfwd.c b/tools/omfwd.c
index 88a382e0..c8fedfc9 100644
--- a/tools/omfwd.c
+++ b/tools/omfwd.c
@@ -82,12 +82,14 @@ typedef struct _instanceData {
 	permittedPeers_t *pPermPeers;
 	int iStrmDrvrMode;
 	char	*f_hname;
-	int *pSockArray;		/* sockets to use for UDP */
+	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 */
+	int compressionLevel;	/* 0 - no compression, else level for zlib */
 	char *port;
 	int protocol;
+	int iUDPRebindInterval;	/* 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 */
@@ -100,9 +102,31 @@ static uchar *pszStrmDrvr = NULL; /* name of the stream driver to use */
 static short iStrmDrvrMode = 0; /* mode for stream driver, driver-dependent (0 mostly means plain tcp) */
 static short bResendLastOnRecon = 0; /* should the last message be re-sent on a successful reconnect? */
 static uchar *pszStrmDrvrAuthMode = NULL; /* authentication mode to use */
+static int iUDPRebindInterval = 0;	/* support for automatic re-binding (load balancers!). 0 - no rebind */
 
 static permittedPeers_t *pPermPeers = NULL;
 
+static rsRetVal doTryResume(instanceData *pData);
+
+/* Close the UDP sockets.
+ * rgerhards, 2009-05-29
+ */
+static rsRetVal
+closeUDPSockets(instanceData *pData)
+{
+	DEFiRet;
+	assert(pData != NULL);
+	if(pData->pSockArray != NULL) {
+		net.closeUDPListenSockets(pData->pSockArray);
+		pData->pSockArray = NULL;
+		freeaddrinfo(pData->f_addr);
+		pData->f_addr = NULL;
+	}
+pData->bIsConnected = 0; // TODO: remove this variable altogether
+	RETiRet;
+}
+
+
 /* get the syslog forward port from selector_t. The passed in
  * struct must be one that is setup for forwarding.
  * rgerhards, 2007-06-28
@@ -148,30 +172,19 @@ ENDisCompatibleWithFeature
 
 BEGINfreeInstance
 CODESTARTfreeInstance
-	if(pData->f_addr != NULL) { /* TODO: is the check ok? */
-		freeaddrinfo(pData->f_addr);
-		pData->f_addr = NULL;
-	}
-	if(pData->port != NULL)
-		free(pData->port);
-
 	/* final cleanup */
 	DestructTCPInstanceData(pData);
-	if(pData->pSockArray != NULL)
-		net.closeUDPListenSockets(pData->pSockArray);
+	closeUDPSockets(pData);
 
 	if(pData->protocol == FORW_TCP) {
 		tcpclt.Destruct(&pData->pTCPClt);
 	}
 
-	if(pData->f_hname != NULL)
-		free(pData->f_hname);
-	if(pData->pszStrmDrvr != NULL)
-		free(pData->pszStrmDrvr);
-	if(pData->pszStrmDrvrAuthMode != NULL)
-		free(pData->pszStrmDrvrAuthMode);
-	if(pData->pPermPeers != NULL)
-		net.DestructPermittedPeers(&pData->pPermPeers);
+	free(pData->port);
+	free(pData->f_hname);
+	free(pData->pszStrmDrvr);
+	free(pData->pszStrmDrvrAuthMode);
+	net.DestructPermittedPeers(&pData->pPermPeers);
 ENDfreeInstance
 
 
@@ -192,6 +205,18 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
 	unsigned lsent = 0;
 	int bSendSuccess;
 
+dbgprintf("rebind logic: interval %d, curr %d, mod %d, if %d\n", pData->iUDPRebindInterval, pData->nXmit,
+	(pData->nXmit % pData->iUDPRebindInterval), ((pData->nXmit % pData->iUDPRebindInterval) == 0));
+	if(pData->iUDPRebindInterval && (pData->nXmit++ % pData->iUDPRebindInterval == 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));
+	}
+
+	if(pData->pSockArray == NULL) {
+		CHKiRet(doTryResume(pData));
+	}
+
 	if(pData->pSockArray != NULL) {
 		/* we need to track if we have success sending to the remote
 		 * peer. Success is indicated by at least one sendto() call
@@ -224,6 +249,7 @@ static rsRetVal UDPSend(instanceData *pData, char *msg, size_t len)
 		}
 	}
 
+finalize_it:
 	RETiRet;
 }
 
@@ -616,6 +642,9 @@ CODE_STD_STRING_REQUESTparseSelectorAct(1)
 		CHKmalloc(pData->f_hname = strdup((char*) q));
 	}
 
+	/* copy over config data as needed */
+	pData->iUDPRebindInterval = iUDPRebindInterval;
+
 	/* process template */
 	CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, OMSR_NO_RQD_TPL_OPTS,
 		(pszTplName == NULL) ? (uchar*)"RSYSLOG_TraditionalForwardFormat" : pszTplName));
@@ -699,6 +728,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a
 	/* we now must reset all non-string values */
 	iStrmDrvrMode = 0;
 	bResendLastOnRecon = 0;
+	iUDPRebindInterval = 0;
 
 	return RS_RET_OK;
 }
@@ -713,6 +743,7 @@ CODEmodInit_QueryRegCFSLineHdlr
 	CHKiRet(objUse(net,LM_NET_FILENAME));
 
 	CHKiRet(regCfSysLineHdlr((uchar *)"actionforwarddefaulttemplate", 0, eCmdHdlrGetWord, NULL, &pszTplName, NULL));
+	CHKiRet(regCfSysLineHdlr((uchar *)"actionsendudprebindinterval", 0, eCmdHdlrInt, NULL, &iUDPRebindInterval, NULL));
 	CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriver", 0, eCmdHdlrGetWord, NULL, &pszStrmDrvr, NULL));
 	CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdrivermode", 0, eCmdHdlrInt, NULL, &iStrmDrvrMode, NULL));
 	CHKiRet(regCfSysLineHdlr((uchar *)"actionsendstreamdriverauthmode", 0, eCmdHdlrGetWord, NULL, &pszStrmDrvrAuthMode, NULL));
diff --git a/tools/syslogd.c b/tools/syslogd.c
index 47fa2f58..3d91e22f 100644
--- a/tools/syslogd.c
+++ b/tools/syslogd.c
@@ -2626,7 +2626,7 @@ init(void)
 		bHadConfigErr = 1;
 	}
 
-	if(localRet != RS_RET_OK || iNbrActions == 0) {
+	if((localRet != RS_RET_OK && localRet != RS_RET_NONFATAL_CONFIG_ERR) || iNbrActions == 0) {
 		/* rgerhards: this code is executed to set defaults when the
 		 * config file could not be opened. We might think about
 		 * abandoning the run in this case - but this, too, is not