From 01c4bfc652004fd7a9a34c0866f821026f09c8fc Mon Sep 17 00:00:00 2001
From: lothar <lothar@aef3b71b-58ee-0310-9ba9-8811b9f0742f>
Date: Mon, 16 Jun 2008 09:23:54 +0000
Subject: [PATCH] - merge changes from /branches/vermont/bloomfilter into
 /trunk

git-svn-id: file:///Users/braun/svn/vermont/trunk/vermont@1729 aef3b71b-58ee-0310-9ba9-8811b9f0742f
---
 CMakeLists.txt                       |  36 ++++
 cmake/modules/FindGSL.cmake          |  30 ++++
 common/AgeBloomFilter.cpp            |  57 +++++++
 common/AgeBloomFilter.h              |  59 +++++++
 common/BloomFilter.cpp               |  96 +++++++++++
 common/BloomFilter.h                 |  73 ++++++++
 common/BloomFilterBase.h             | 239 +++++++++++++++++++++++++++
 common/CMakeLists.txt                |   3 +
 common/CountBloomFilter.cpp          |  42 +++++
 common/CountBloomFilter.h            |  34 ++++
 common/InstanceManager.h             |   8 +-
 common/MinBloomFilter.h              |  59 +++++++
 common/StatisticsManager.cpp         |   3 +-
 common/StatisticsManager.h           |   7 +-
 concentrator/IpfixSender.cpp         |   4 +
 configs/connectionfilter.xml         |  72 ++++++++
 configs/stateconnectionfilter.xml    |  70 ++++++++
 ipfix_configuration.cc               |  12 +-
 ipfix_configuration.h                |  15 +-
 metering_configuration.cc            |  16 ++
 metering_configuration.h             |   2 +
 observer_configuration.cc            |   1 +
 packetselection_configuration.cc     |  40 +++++
 pcapexporter_configuration.cc        |  59 +++++++
 pcapexporter_configuration.h         |  35 ++++
 sampler/CMakeLists.txt               |   3 +
 sampler/ConnectionFilter.cpp         |  84 ++++++++++
 sampler/ConnectionFilter.h           |  29 ++++
 sampler/Observer.cpp                 |  26 ++-
 sampler/Observer.h                   |   3 +-
 sampler/Packet.h                     |   5 +-
 sampler/PacketReceiver.cpp           |   4 +-
 sampler/PacketReceiver.h             |   7 +-
 sampler/PcapExporterSink.cpp         |  49 ++++++
 sampler/PcapExporterSink.h           |  54 ++++++
 sampler/Sink.h                       |   2 +-
 sampler/StateConnectionFilter.cpp    |  60 +++++++
 sampler/StateConnectionFilter.h      |  24 +++
 test_build.sh                        |   2 +
 tests/AggregationPerfTest.cpp        |   4 +-
 tests/AggregationPerfTest.h          |   2 +-
 tests/BloomFilterTest.cpp            | 139 ++++++++++++++++
 tests/BloomFilterTest.h              |  20 +++
 tests/CMakeLists.txt                 |   8 +-
 tests/ConnectionFilterTest.cpp       | 110 ++++++++++++
 tests/ConnectionFilterTest.h         |  19 +++
 tests/Makefile.am                    |  32 ----
 tests/VermontTest.cpp                |  11 +-
 tests/data/connectionfiltertest.pcap | Bin 0 -> 1907 bytes
 tests/test_concentrator.cpp          |   2 +-
 tools/Makefile.am                    |   9 -
 51 files changed, 1695 insertions(+), 85 deletions(-)
 create mode 100644 cmake/modules/FindGSL.cmake
 create mode 100644 common/AgeBloomFilter.cpp
 create mode 100644 common/AgeBloomFilter.h
 create mode 100644 common/BloomFilter.cpp
 create mode 100644 common/BloomFilter.h
 create mode 100644 common/BloomFilterBase.h
 create mode 100644 common/CountBloomFilter.cpp
 create mode 100644 common/CountBloomFilter.h
 create mode 100644 common/MinBloomFilter.h
 create mode 100644 configs/connectionfilter.xml
 create mode 100644 configs/stateconnectionfilter.xml
 create mode 100644 pcapexporter_configuration.cc
 create mode 100644 pcapexporter_configuration.h
 create mode 100644 sampler/ConnectionFilter.cpp
 create mode 100644 sampler/ConnectionFilter.h
 create mode 100644 sampler/PcapExporterSink.cpp
 create mode 100644 sampler/PcapExporterSink.h
 create mode 100644 sampler/StateConnectionFilter.cpp
 create mode 100644 sampler/StateConnectionFilter.h
 create mode 100644 tests/BloomFilterTest.cpp
 create mode 100644 tests/BloomFilterTest.h
 create mode 100644 tests/ConnectionFilterTest.cpp
 create mode 100644 tests/ConnectionFilterTest.h
 delete mode 100644 tests/Makefile.am
 create mode 100644 tests/data/connectionfiltertest.pcap
 delete mode 100644 tools/Makefile.am

diff --git a/CMakeLists.txt b/CMakeLists.txt
index b9ede05..4c4b84d 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -43,6 +43,7 @@ SUBDIRS(common concentrator ipfixlolib sampler)
 ADD_EXECUTABLE(vermont
 	collector_configuration.cc
 	exporter_configuration.cc
+	pcapexporter_configuration.cc
 	metering_configuration.cc
 	observer_configuration.cc
 	vermont.cc
@@ -159,9 +160,11 @@ IF (DEBUG)
 	message(STATUS "Configuring build process for debug version")
 	REMOVE_DEFINITIONS(-O3)
 	ADD_DEFINITIONS(-O0 -g -pg -Wall -Werror -DDEBUG)
+	SET_TARGET_PROPERTIES(vermont PROPERTIES LINK_FLAGS "-g -pg")
 ELSE (DEBUG)
 	REMOVE_DEFINITIONS(-O0 -g -pg -Wall -Werror -DDEBUG)
 	ADD_DEFINITIONS(-O3)
+	SET_TARGET_PROPERTIES(vermont PROPERTIES LINK_FLAGS "")
 ENDIF (DEBUG)
 
 
@@ -253,6 +256,39 @@ ELSE (SUPPORT_SCTP)
     REMOVE_DEFINITIONS(-DSUPPORT_SCTP)
 ENDIF (SUPPORT_SCTP)
 
+### connection filter
+
+OPTION(CONNECTION_FILTER "Enables/disables the connection filter." ON)
+
+IF (CONNECTION_FILTER)
+	ADD_DEFINITIONS(-DHAVE_CONNECTION_FILTER)
+ENDIF(CONNECTION_FILTER)
+
+### gsl
+
+OPTION(USE_GSL "Enables/disables GSL in connectionflter." ON)
+IF (CONNECTION_FILTER AND NOT USE_GSL)
+ 	MESSAGE(FATAL_ERROR "GSL is needed for Connectionfilter at the moment.
+	You cannot have -DCONNECTION_FILTER=ON and -DUSE_GSL=OFF")
+ENDIF (CONNECTION_FILTER AND NOT USE_GSL)
+
+IF (USE_GSL)
+	FIND_PACKAGE(GSL)
+	MARK_AS_ADVANCED(
+		GSL_INCLUDE_DIR
+		GSL_LIBRARIES
+		GSL_LIBRARY
+		BLAS_LIBRARY
+	)	
+	IF (GSL_FOUND)
+		MESSAGE(STATUS "GNU scientific library found")
+	ELSE (GSL_FOUND)
+		MESSAGE(FATAL_ERROR "GNU scientific library not found. Please
+			install the library or use -DCONNECTION_FILTER=OFF")
+	ENDIF (GSL_FOUND)
+	ADD_DEFINITIONS(-DHAVE_GSL)
+	TARGET_LINK_LIBRARIES(vermont ${GSL_LIBRARIES})
+ENDIF (USE_GSL)
 
 ### tools
 
diff --git a/cmake/modules/FindGSL.cmake b/cmake/modules/FindGSL.cmake
new file mode 100644
index 0000000..5f2c308
--- /dev/null
+++ b/cmake/modules/FindGSL.cmake
@@ -0,0 +1,30 @@
+#################################### Locate gsl
+FIND_PATH(
+  GSL_INCLUDE_DIR
+  gsl/gsl_cdf.h
+  /usr/include/ /usr/include/gsl/
+  /use/local/include/ /usr/local/include/gsl/
+  )
+
+FIND_LIBRARY(
+  GSL_LIBRARY
+  NAMES gsl
+  PATHS /usr/lib /usr/local/lib
+  PATH_SUFFIXES gsl
+  )
+
+FIND_LIBRARY(
+  BLAS_LIBRARY
+  NAMES gslcblas blas cblas
+  PATHS /usr/lib /usr/local/lib
+  PATH_SUFFIXES gsl blas cblas
+  )
+
+IF (GSL_LIBRARY AND BLAS_LIBRARY)
+  SET(GSL_LIBRARIES ${GSL_LIBRARY} ${BLAS_LIBRARY})
+ENDIF (GSL_LIBRARY AND BLAS_LIBRARY)
+
+IF (GSL_INCLUDE_DIR AND GSL_LIBRARIES)
+  SET(GSL_FOUND TRUE)
+ENDIF (GSL_INCLUDE_DIR AND GSL_LIBRARIES)
+
diff --git a/common/AgeBloomFilter.cpp b/common/AgeBloomFilter.cpp
new file mode 100644
index 0000000..acfebb7
--- /dev/null
+++ b/common/AgeBloomFilter.cpp
@@ -0,0 +1,57 @@
+/**************************************************************************/
+/*    Copyright (C) 2007 Gerhard Muenz                                    */
+/*                                                                        */
+/*    This 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 2.1 of the License, or (at your option) any later version.  */
+/*                                                                        */
+/*    This 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 this library; if not, write to the Free Software  */
+/*    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA    */
+/**************************************************************************/
+
+#include "AgeBloomFilter.h"
+
+void AgeArray::resize(uint32_t size)
+{
+	free(array);
+	array_size = size;
+	array = (agetime_t*)malloc(size*sizeof(agetime_t));
+	clear();
+}
+
+void AgeArray::clear()
+{
+	memset(array, 0, array_size*sizeof(agetime_t));
+}
+
+void AgeArray::set(size_t index, agetime_t time)
+{
+	if(index < array_size) {
+		array[index] = time;
+	}
+}
+
+agetime_t AgeArray::get(size_t index) const
+{
+    if(index < array_size)
+	return array[index];
+    else
+	return 0;
+}
+
+std::ostream & operator << (std::ostream & os, const AgeArray & a) 
+{
+    for(uint32_t i=0; i<a.array_size; i++)
+    {
+	os << a.get(i) << " ";
+    }
+    return os;
+}
+
diff --git a/common/AgeBloomFilter.h b/common/AgeBloomFilter.h
new file mode 100644
index 0000000..b412a11
--- /dev/null
+++ b/common/AgeBloomFilter.h
@@ -0,0 +1,59 @@
+/**************************************************************************/
+/*    Copyright (C) 2007 Gerhard Muenz                                    */
+/*                                                                        */
+/*    This 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 2.1 of the License, or (at your option) any later version.  */
+/*                                                                        */
+/*    This 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 this library; if not, write to the Free Software  */
+/*    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA    */
+/**************************************************************************/
+
+#ifndef _AGE_BLOOMFILTER_H_
+#define _AGE_BLOOMFILTER_H_
+
+#include "MinBloomFilter.h"
+
+#include <ostream>
+
+typedef time_t agetime_t;
+
+class AgeArray {
+	friend std::ostream & operator << (std::ostream &, const AgeArray &);
+
+	public:
+		AgeArray(size_t size = 0) : array(NULL)
+		{
+			resize(size);
+		}
+	
+		~AgeArray()
+		{
+			free(array);
+		}
+
+		typedef agetime_t ValueType;
+
+		void resize(size_t size);
+		void clear();
+		void set(size_t index, agetime_t time);
+		agetime_t get(size_t index) const;
+    
+	private:
+		agetime_t* array;
+		size_t array_size;
+};
+
+std::ostream & operator << (std::ostream &, const AgeArray &);
+
+
+typedef MinBloomFilter<AgeArray> AgeBloomFilter;
+
+#endif // _AGE_BLOOMFILTER_H_
diff --git a/common/BloomFilter.cpp b/common/BloomFilter.cpp
new file mode 100644
index 0000000..04c3976
--- /dev/null
+++ b/common/BloomFilter.cpp
@@ -0,0 +1,96 @@
+/**************************************************************************/
+/*    Copyright (C) 2007 Gerhard Muenz                                    */
+/*                                                                        */
+/*    This 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 2.1 of the License, or (at your option) any later version.  */
+/*                                                                        */
+/*    This 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 this library; if not, write to the Free Software  */
+/*    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA    */
+/**************************************************************************/
+
+#include "BloomFilter.h"
+
+const uint8_t bitmask[8] =
+{
+    0x01, //00000001
+    0x02, //00000010
+    0x04, //00000100
+    0x08, //00001000
+    0x10, //00010000
+    0x20, //00100000
+    0x40, //01000000
+    0x80  //10000000
+};
+    
+void Bitmap::resize(size_t size)
+{
+    free(bitmap);
+    len_bits = size;
+    len_octets = (size+7)/8;
+    bitmap = (uint8_t*)(malloc(len_octets));
+    memset(bitmap, 0, len_octets);
+}
+
+void Bitmap::clear()
+{
+    memset(bitmap, 0, len_octets);
+}
+
+void Bitmap::set(size_t index)
+{
+    if(index < len_bits)
+	bitmap[index/8] |= bitmask[index%8];
+}
+
+bool Bitmap::get(size_t index) const
+{
+    if(index < len_bits)
+	return (bitmap[index/8] & bitmask[index%8]) != 0;
+    else
+	return false;
+}
+
+std::ostream & operator<< (std::ostream & os, const Bitmap & b) 
+{
+    for(size_t i=0; i<b.len_bits; i++)
+    {
+	if(b.get(i))
+	    os << '1';
+	else
+	    os << '0';
+    }
+    return os;
+}
+
+void BloomFilter::set(const uint8_t* input, size_t len, bool) 
+{
+    for(unsigned i=0; i < hfList->len; i++) {
+    	if (CMS_)
+            filter_[0].set(hashU(input, len, filterSize(), hfList->seed[i]));
+	else 
+	    filter_[i].set(hashU(input, len, filterSize(), hfList->seed[i]));
+    }
+}
+
+bool BloomFilter::get(const uint8_t* input, size_t len) const
+{
+    for(unsigned i=0; i < hfList->len; i++) {
+        if (CMS_) {
+	    if (filter_[0].get(hashU(input, len, filterSize(), hfList->seed[i])) == false)
+	        return false;
+        } else {
+            if (filter_[i].get(hashU(input, len, filterSize(), hfList->seed[i])) == false)
+	        return false;
+	}
+    }
+    return true;
+}
+
diff --git a/common/BloomFilter.h b/common/BloomFilter.h
new file mode 100644
index 0000000..ec8f758
--- /dev/null
+++ b/common/BloomFilter.h
@@ -0,0 +1,73 @@
+/**************************************************************************/
+/*    Copyright (C) 2007 Gerhard Muenz                                    */
+/*                                                                        */
+/*    This 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 2.1 of the License, or (at your option) any later version.  */
+/*                                                                        */
+/*    This 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 this library; if not, write to the Free Software  */
+/*    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA    */
+/**************************************************************************/
+
+#ifndef _BLOOMFILTER_H_
+#define _BLOOMFILTER_H_
+
+#include "BloomFilterBase.h"
+
+#include <ostream>
+
+/* Bitmap vector for normal BloomFilter */
+class Bitmap {
+    friend std::ostream & operator << (std::ostream &, const Bitmap &);
+
+    public:
+	Bitmap(size_t size = 0) : bitmap(NULL)
+	{
+	    resize(size);
+	}
+
+
+	~Bitmap()
+	{
+	    free(bitmap);
+	}
+
+	typedef bool ValueType;
+
+	void resize(size_t size);
+	void clear();
+	void set(size_t index);
+	bool get(size_t index) const;
+
+    private:
+	uint8_t* bitmap;
+	size_t len_bits;
+	size_t len_octets;
+};
+
+std::ostream & operator << (std::ostream &, const Bitmap &);
+
+
+
+/* BloomFilter normal bloom filter */
+class BloomFilter : public BloomFilterBase<Bitmap>
+{
+    friend std::ostream & operator << (std::ostream &, const BloomFilter &);
+
+    public:
+	BloomFilter(HashParams* hashParams, size_t size, bool CMS = true) : BloomFilterBase<Bitmap>(hashParams, size, CMS) {}
+
+	virtual ~BloomFilter() {}
+
+	virtual bool get(const uint8_t* input, size_t len) const;
+	virtual void set(const uint8_t* input, size_t len, bool v = true);
+};
+
+#endif // _BLOOMFILTER_H_
diff --git a/common/BloomFilterBase.h b/common/BloomFilterBase.h
new file mode 100644
index 0000000..66c038e
--- /dev/null
+++ b/common/BloomFilterBase.h
@@ -0,0 +1,239 @@
+#ifndef _BLOOMFILTER_BASE_H_
+#define _BLOOMFILTER_BASE_H_
+
+#ifdef HAVE_GSL
+#include <gsl/gsl_rng.h>
+#endif
+
+#include <stdint.h>
+#include <cstring>
+#include <ctime>
+
+#include <sampler/Packet.h>
+
+/* GenericKey class holding uint8_t* input for BloomFilter hash functions */
+template<unsigned size> class GenericKey
+{
+	public:
+		unsigned len;
+		uint8_t data[size];
+
+		GenericKey() : len(0) {
+			memset(data,0,sizeof(data));
+		}
+
+		void set(uint8_t *input, unsigned inputlen)
+		{
+			len = inputlen<sizeof(data)?inputlen:sizeof(data);
+			memcpy(&data, input, len);
+		}
+
+		void reset()
+		{
+			len = 0;
+			memset(data,0,sizeof(data));
+		}
+
+		void append(uint8_t *input, unsigned inputlen)
+		{
+			if(len<sizeof(data)) {
+				if((len+inputlen) < sizeof(data)) {
+					memcpy(&(data[len]), input, inputlen);
+					len = len + inputlen;
+				} else {
+					memcpy(&(data[len]), input, sizeof(data)-len);
+					len = sizeof(data);
+				}
+			}
+		}
+
+		bool operator<(const GenericKey& other) const 
+		{
+			if(len < other.len)
+				return true;
+			else if(len > other.len)
+				return false;
+			else
+				return memcmp(data, other.data, len)<0?true:false;
+		}
+};
+
+
+/* QuintupleKey class holding input for BloomFilter hash functions */
+class QuintupleKey
+{
+	public:
+		uint8_t data[15];
+		const unsigned len;
+
+		struct Quintuple
+		{
+			uint32_t srcIp;
+			uint32_t dstIp;
+			uint8_t proto;
+			uint16_t srcPort;
+			uint16_t dstPort;
+		};
+
+		QuintupleKey() : len(15)
+		{
+			memset(data,0,sizeof(data));
+		}
+
+		QuintupleKey(const Packet* p) : len(15) 
+		{
+			memset(data,0, sizeof(data));
+			uint32_t ip1, ip2;
+			uint16_t port1, port2;
+			if (p->ipProtocolType == Packet::TCP) {
+				ip1 = *((uint32_t*)(p->netHeader + 12));
+				ip2 = *((uint32_t*)(p->netHeader + 16));
+				port1 = *((uint16_t*)(p->transportHeader));
+				port2 = *((uint16_t*)(p->transportHeader + 2));
+
+				getQuintuple()->srcIp   = ip1>ip2?ip1:ip2;
+				getQuintuple()->dstIp   = ip1>ip2?ip2:ip1;
+				getQuintuple()->proto   = p->ipProtocolType;
+				getQuintuple()->srcPort = port1>port2?port1:port2;
+				getQuintuple()->dstPort = port1>port2?port2:port1;
+			}
+		}
+
+		inline Quintuple* getQuintuple()
+		{
+			return (Quintuple*)data;
+		}
+
+		void reset()
+		{
+			memset(data,0,sizeof(data));
+		}
+
+		bool operator<(const QuintupleKey& other) const 
+		{
+			return memcmp(data, other.data, len)<0?true:false;
+		}
+};
+
+struct HashParams
+{
+	HashParams(size_t l, unsigned startSeed = time(0)) : len(l) {
+		seed = (uint32_t*)calloc(sizeof(uint32_t), len);
+		srand(startSeed);
+		for(unsigned i=0; i < len; i++) {
+			seed[i] = rand();
+		}
+	}
+
+	~HashParams() {
+		free(seed);
+	}
+	size_t len;
+	uint32_t* seed;
+};
+
+/**
+ * BloomFilterBase provides hash functions for filters and is the interface for all type of 
+ * BloomFilters. The class needs a template parameter which has to provide the following
+ * methods/types:
+ *
+ * class MyT {
+ * public:
+ * 	MyT(size_t size);
+ *	typedef ValueType myDesiredType;
+ *	void clear();
+ *	myDesiredType get(uint8_t* data, size_t len);
+ *	void set(uint8_t* data, size_t len, myDesiredType value);
+ * }
+ */
+template <class T>
+class BloomFilterBase
+{
+	public:
+		BloomFilterBase(HashParams* hashParams, unsigned filterSize, bool CMS = true) : hfList(hashParams),
+			filterSize_(filterSize), CMS_(CMS)
+		{
+#ifdef HAVE_GSL
+			r = gsl_rng_alloc (gsl_rng_fishman18);
+#endif
+			if (CMS_) {
+				filterCount_ = 1;
+			} else {
+				filterCount_ = hashParams->len;
+			}
+
+			filter_ = new T[filterCount_];
+			for (unsigned i = 0; i != filterCount_; ++i) {
+				filter_[i].resize(filterSize_);
+			}
+		
+			clear();
+		}
+
+		virtual ~BloomFilterBase()
+		{
+#ifdef HAVE_GSL
+			gsl_rng_free(r);
+#endif
+			delete[] filter_;
+		}
+
+		virtual typename T::ValueType get(const uint8_t* data, size_t len) const = 0;
+		virtual void set(const uint8_t* data, size_t len, typename T::ValueType) = 0;
+
+		size_t filterSize() const {
+			return filterSize_;
+		}
+
+
+	protected:
+
+#ifdef HAVE_GSL
+		uint32_t hashU(const uint8_t* input, uint16_t len, uint32_t max, uint32_t seed) const
+		{
+			uint32_t result = 0;
+			gsl_rng_set(r, seed);
+			for (unsigned i = 0; i < len; i++) {
+				result += input[i] * gsl_rng_get(r);
+			}
+			return result % max;
+		}
+#else
+		uint32_t hashU(const uint8_t* input, uint16_t len, uint32_t max, uint32_t seed) const
+		{
+			// TODO: create a function that does not need gsl
+			return 0;
+		}
+#endif
+
+		int32_t ggT(uint32_t m, uint32_t n)
+		{
+			uint32_t z;
+			while (n>0) {
+				z=n;
+				n=m%n;
+				m=z;
+			}
+			return m;
+		}
+
+		void clear()
+		{
+			for (unsigned i = 0; i != filterCount_; ++i) {
+				filter_[i].clear();
+			}
+		}
+
+#ifdef HAVE_GSL
+		gsl_rng * r;
+#endif
+
+		const HashParams* hfList;
+
+		size_t filterSize_;
+		T* filter_;
+		size_t filterCount_;
+		bool CMS_;
+};
+
+#endif
diff --git a/common/CMakeLists.txt b/common/CMakeLists.txt
index 7d0c49d..41c4d6c 100644
--- a/common/CMakeLists.txt
+++ b/common/CMakeLists.txt
@@ -21,5 +21,8 @@ ADD_LIBRARY(common
 	TimeoutSemaphore.cpp
 	msg.cc
 	StatisticsManager.cpp
+	BloomFilter.cpp
+	AgeBloomFilter.cpp
+	CountBloomFilter.cpp
 )
 
diff --git a/common/CountBloomFilter.cpp b/common/CountBloomFilter.cpp
new file mode 100644
index 0000000..6928120
--- /dev/null
+++ b/common/CountBloomFilter.cpp
@@ -0,0 +1,42 @@
+#include "CountBloomFilter.h"
+
+void CountArray::resize(size_t size)
+{
+	free(array);
+	array_size = size;
+	array = (ValueType*)(malloc(size*sizeof(ValueType)));
+	clear();
+}
+
+void CountArray::clear()
+{
+	memset(array, 0, array_size*sizeof(ValueType));
+}
+
+void CountArray::set(size_t index, ValueType value)
+{
+	if(index < array_size) {
+		array[index] += value;
+		//msg(MSG_DEBUG, "array[%u] == %u", index, array[index]);
+	}
+}
+
+CountArray::ValueType CountArray::get(size_t index) const
+{
+	if(index < array_size) {
+		//msg(MSG_DEBUG, "get: array[%u] == %u", index, array[index]);
+		return array[index];
+	} else {
+		return 0;
+	}
+}
+
+std::ostream & operator << (std::ostream & os, const CountArray & a) 
+{
+    for(size_t i=0; i<a.array_size; i++)
+    {
+	os << a.get(i) << " ";
+    }
+    return os;
+}
+
diff --git a/common/CountBloomFilter.h b/common/CountBloomFilter.h
new file mode 100644
index 0000000..bd68eed
--- /dev/null
+++ b/common/CountBloomFilter.h
@@ -0,0 +1,34 @@
+#ifndef _COUNT_BLOOMFILTER_H_
+#define _COUNT_BLOOMFILTER_H_
+
+#include "MinBloomFilter.h"
+
+
+#include <ostream>
+
+class CountArray {
+	public:
+		CountArray(size_t size = 0) : array(NULL) {
+			resize(size);
+		}
+
+		~CountArray() {
+			free(array);
+		}
+
+		typedef int ValueType;
+
+		void resize(size_t size);
+		void clear();
+		void set(size_t index, ValueType value);
+		ValueType get(size_t index) const;
+
+	private:
+		ValueType* array;
+		size_t array_size;
+		friend std::ostream & operator << (std::ostream &, const CountArray &);
+};
+
+typedef MinBloomFilter<CountArray> CountBloomFilter;
+
+#endif
diff --git a/common/InstanceManager.h b/common/InstanceManager.h
index 5ce90f5..4f68605 100644
--- a/common/InstanceManager.h
+++ b/common/InstanceManager.h
@@ -26,8 +26,6 @@
 #include <queue>
 #include <list>
 
-using namespace std;
-
 /**
  * manages instances of the given type to avoid news/deletes in program
  * managed types *should* be inherited from ManagedInstance
@@ -38,9 +36,9 @@ class InstanceManager
 {
 	private:
 #if defined(DEBUG)
-		list<T*> usedInstances;	// instances with active references (only used for debugging purposes)
+		std::list<T*> usedInstances;	// instances with active references (only used for debugging purposes)
 #endif
-		queue<T*> freeInstances;// unused instances
+		std::queue<T*> freeInstances;// unused instances
 		Mutex mutex;			// we wanna be thread-safe
 		static const int DEFAULT_NO_INSTANCES = 1000;
 
@@ -124,7 +122,7 @@ class InstanceManager
 				mutex.lock();
 				freeInstances.push(instance);
 #if defined(DEBUG)
-				typename list<T*>::iterator iter = find(usedInstances.begin(), usedInstances.end(), instance);
+				typename std::list<T*>::iterator iter = find(usedInstances.begin(), usedInstances.end(), instance);
 				if (iter == usedInstances.end()) {
 					THROWEXCEPTION("instance (0x%08X) is not managed by InstanceManager", (void*)instance);
 				}
diff --git a/common/MinBloomFilter.h b/common/MinBloomFilter.h
new file mode 100644
index 0000000..c703874
--- /dev/null
+++ b/common/MinBloomFilter.h
@@ -0,0 +1,59 @@
+#ifndef _MIN_BLOOMFILTER_
+#define _MIN_BLOOMFILTER_
+
+#include "BloomFilterBase.h"
+
+#include "msg.h"
+
+#include <climits>
+
+// work around a gcc "feature": whenever you need to access a variable from BloomFilterBase<T>,
+// you need to specify the full namespace, e.g.  BloomFilterBase::hf_numbers
+
+template <class T>
+class MinBloomFilter : public BloomFilterBase<T>
+{
+	public:
+		MinBloomFilter(HashParams* hashParams, size_t filterSize, bool CMS = true)
+			: BloomFilterBase<T>(hashParams, filterSize, CMS) {}
+
+		virtual ~MinBloomFilter() {}
+
+
+		virtual typename T::ValueType  get(const uint8_t* input, size_t len) const {
+			typename T::ValueType  ret = INT_MAX;
+			typename T::ValueType  current;
+			for(unsigned i=0; i != BloomFilterBase<T>::hfList->len; i++) {   
+				if (BloomFilterBase<T>::CMS_) {
+					current = BloomFilterBase<T>::filter_[0].get(
+						BloomFilterBase<T>::hashU(input, len, 
+							BloomFilterBase<T>::filterSize(), BloomFilterBase<T>::hfList->seed[i]));
+				} else {
+					current = BloomFilterBase<T>::filter_[i].get(
+						BloomFilterBase<T>::hashU(input, len,
+							BloomFilterBase<T>::filterSize(), BloomFilterBase<T>::hfList->seed[i]));
+				}
+				if (current < ret)
+					ret = current;
+			}
+			//msg(MSG_DEBUG, "MinBloomFilter.get(): %i", ret);
+			return ret;
+		}
+
+		virtual void set(const uint8_t* input, size_t len, typename T::ValueType v) {
+			//msg(MSG_DEBUG, "MinBloomFilter.set(): %i", v);
+			for(unsigned i=0; i != BloomFilterBase<T>::hfList->len; i++) {
+				if (BloomFilterBase<T>::CMS_) {
+					BloomFilterBase<T>::filter_[0].set(BloomFilterBase<T>::hashU(input, len,
+						BloomFilterBase<T>::filterSize(), 
+						BloomFilterBase<T>::hfList->seed[i]), v);
+				} else {
+					BloomFilterBase<T>::filter_[i].set(BloomFilterBase<T>::hashU(input, len,
+						BloomFilterBase<T>::filterSize(), 
+						BloomFilterBase<T>::hfList->seed[i]), v);
+				}
+			}
+		}
+};
+
+#endif
diff --git a/common/StatisticsManager.cpp b/common/StatisticsManager.cpp
index 7ab4bf0..93ba2e9 100644
--- a/common/StatisticsManager.cpp
+++ b/common/StatisticsManager.cpp
@@ -4,6 +4,7 @@
 #include <string>
 #include <unistd.h>
 
+using namespace std;
 
 StatisticsManager::StatisticsManager()
 	: Thread(threadWrapper), interval(10000)
@@ -67,7 +68,7 @@ void StatisticsManager::runStats()
 
 		FILE* f = fopen(outputFile.c_str(), "a");
 		if (f == 0) THROWEXCEPTION("failed to open file %s", outputFile.c_str());
-		fprintf(f, "statistics dump at %lu\n", time(0));
+		fprintf(f, "statistics dump at %lu\n", (long unsigned)time(0));
 
 		mutex.lock();
 		list<StatisticsModule*>::const_iterator iter = statModules.begin();
diff --git a/common/StatisticsManager.h b/common/StatisticsManager.h
index a0db5ca..0f37b91 100644
--- a/common/StatisticsManager.h
+++ b/common/StatisticsManager.h
@@ -7,7 +7,6 @@
 #include <list>
 #include <string>
 
-using namespace std;
 
 // statistics output
 class StatisticsModule
@@ -20,10 +19,10 @@ class StatisticsModule
 class StatisticsManager : Thread
 {
 	private:	
-		list<StatisticsModule*> statModules;
+		std::list<StatisticsModule*> statModules;
 		unsigned long			interval;
 		Mutex					mutex;
-		string					outputFile;
+		std::string					outputFile;
 
 		StatisticsManager();
 		static void* threadWrapper(void* sm);
@@ -37,7 +36,7 @@ class StatisticsManager : Thread
 		void start();
 		void stop();
 		void setInterval(long milliseconds);
-		void setOutput(const string& output);
+		void setOutput(const std::string& output);
 
 };
 
diff --git a/concentrator/IpfixSender.cpp b/concentrator/IpfixSender.cpp
index a398fa2..a9e08c0 100644
--- a/concentrator/IpfixSender.cpp
+++ b/concentrator/IpfixSender.cpp
@@ -112,9 +112,13 @@ int IpfixSender::addCollector(const char *ip, uint16_t port, ipfix_transport_pro
 	    case SCTP:
 		msg(MSG_INFO, "IpfixSender: adding SCTP://%s:%d to exporter", ip, port);
 		break;
+#ifdef IPFIXLOLIB_RAWDIR_SUPPORT
 	    case RAWDIR:
 		msg(MSG_INFO, "IpfixSender: adding RAWDIR://%s to exporter", ip);
 		break;
+#endif
+	    case TCP:
+	        msg(MSG_INFO, "IpfixSender: adding TCP://%s:%d to exporter", ip, port);
 	}
 
 	if(ipfix_add_collector(ex, ip, port, proto) != 0) {
diff --git a/configs/connectionfilter.xml b/configs/connectionfilter.xml
new file mode 100644
index 0000000..795d4c2
--- /dev/null
+++ b/configs/connectionfilter.xml
@@ -0,0 +1,72 @@
+<ipfixConfig xmlns="urn:ietf:params:xml:ns:ipfix-config">
+
+    <observationPoint id="1">
+	<observationDomainId>4711</observationDomainId>
+	<type>pcap</type>
+	<parameters>
+		<interface>eth1</interface>
+		<pcap_filter>ip</pcap_filter>
+	</parameters>
+	<next>
+	    <meteringProcessId>1</meteringProcessId>
+	</next>
+    </observationPoint>
+    <meteringProcess id="1">
+	<packetSelection>
+	    <connectionFilter>
+              <exportBytes>100</exportBytes>
+	      <timeout>3</timeout>
+	      <filterSize>1000</filterSize>
+	      <hashFunctions>3</hashFunctions>
+	    </connectionFilter>
+	</packetSelection>
+	<packetReporting>
+	    <templateId>888</templateId>
+	    <reportedIE>
+		<ieName>sourceIPv4Address</ieName>
+	    </reportedIE>
+	    <reportedIE>
+		<ieName>destinationIPv4Address</ieName>
+	    </reportedIE>
+	    <reportedIE>
+		<ieName>ipPayloadPacketSection</ieName>
+		<ieLength>65535</ieLength>
+		</reportedIE>
+	    <reportedIE>
+		<ieName>protocolIdentifier</ieName>
+	    </reportedIE>
+	</packetReporting>
+	<next>
+	    <exportingProcessId>1</exportingProcessId>
+	</next>
+    </meteringProcess>
+
+    <exportingProcess id="1">
+	<ipfixPacketRestrictions>
+	    <maxPacketSize>1500</maxPacketSize>
+	    <maxExportDelay unit="msec">500</maxExportDelay>
+	</ipfixPacketRestrictions>
+	<udpTemplateManagement>
+	    <templateRefreshTimeout unit="sec">5</templateRefreshTimeout>
+	    <templateRefreshRate>100</templateRefreshRate>
+	</udpTemplateManagement>
+	<collector>
+	    <ipAddressType>4</ipAddressType>
+	    <ipAddress>127.0.0.1</ipAddress>
+	    <transportProtocol>17</transportProtocol>
+	    <port>1500</port>
+	</collector>
+    </exportingProcess>
+
+    
+    <vermont_main>
+	<poll_interval unit="msec">500</poll_interval>
+	<log_file>log.stat</log_file>
+	<log_interval unit="msec">300000</log_interval>
+    </vermont_main>
+    
+</ipfixConfig>
+		    
+			
+		
+		
diff --git a/configs/stateconnectionfilter.xml b/configs/stateconnectionfilter.xml
new file mode 100644
index 0000000..e1b8d55
--- /dev/null
+++ b/configs/stateconnectionfilter.xml
@@ -0,0 +1,70 @@
+<ipfixConfig xmlns="urn:ietf:params:xml:ns:ipfix-config">
+
+    <observationPoint id="1">
+	<observationDomainId>4711</observationDomainId>
+	<type>pcap</type>
+	<parameters>
+		<interface>eth1</interface>
+		<pcap_filter>ip</pcap_filter>
+	</parameters>
+	<next>
+	    <meteringProcessId>1</meteringProcessId>
+	</next>
+    </observationPoint>
+    <meteringProcess id="1">
+	<packetSelection>
+	    <stateConnectionFilter>
+              <exportBytes>100</exportBytes>
+	      <timeout>3</timeout>
+	    </stateConnectionFilter>
+	</packetSelection>
+	<packetReporting>
+	    <templateId>888</templateId>
+	    <reportedIE>
+		<ieName>sourceIPv4Address</ieName>
+	    </reportedIE>
+	    <reportedIE>
+		<ieName>destinationIPv4Address</ieName>
+	    </reportedIE>
+	    <reportedIE>
+		<ieName>ipPayloadPacketSection</ieName>
+		<ieLength>65535</ieLength>
+		</reportedIE>
+	    <reportedIE>
+		<ieName>protocolIdentifier</ieName>
+	    </reportedIE>
+	</packetReporting>
+	<next>
+	    <exportingProcessId>1</exportingProcessId>
+	</next>
+    </meteringProcess>
+
+    <exportingProcess id="1">
+	<ipfixPacketRestrictions>
+	    <maxPacketSize>1500</maxPacketSize>
+	    <maxExportDelay unit="msec">500</maxExportDelay>
+	</ipfixPacketRestrictions>
+	<udpTemplateManagement>
+	    <templateRefreshTimeout unit="sec">5</templateRefreshTimeout>
+	    <templateRefreshRate>100</templateRefreshRate>
+	</udpTemplateManagement>
+	<collector>
+	    <ipAddressType>4</ipAddressType>
+	    <ipAddress>127.0.0.1</ipAddress>
+	    <transportProtocol>17</transportProtocol>
+	    <port>1500</port>
+	</collector>
+    </exportingProcess>
+
+    
+    <vermont_main>
+	<poll_interval unit="msec">500</poll_interval>
+	<log_file>log.stat</log_file>
+	<log_interval unit="msec">300000</log_interval>
+    </vermont_main>
+    
+</ipfixConfig>
+		    
+			
+		
+		
diff --git a/ipfix_configuration.cc b/ipfix_configuration.cc
index a5ab588..f91fdbb 100644
--- a/ipfix_configuration.cc
+++ b/ipfix_configuration.cc
@@ -3,6 +3,7 @@
 #include "metering_configuration.h"
 #include "collector_configuration.h"
 #include "exporter_configuration.h"
+#include "pcapexporter_configuration.h"
 #include "flowmetering_configuration.h"
 #include "vermontmain_configuration.h"
 #include "dbwriter_configuration.h"
@@ -50,7 +51,9 @@ void Configuration::fillNextVector(xmlNodePtr p)
 		} else if (tagMatches(j, "dbWriterId")) {
 			nextVector.push_back(configTypes::dbwriter +
 					     getContent(j));
-		} 
+		} else if (tagMatches(j, "pcapExporterId")) {
+			nextVector.push_back(configTypes::pcapExporter + getContent(j));
+		}
 		j = j->next;
 	}
 
@@ -115,6 +118,8 @@ IpfixConfiguration::IpfixConfiguration(const std::string& configFile)
 			conf = new MeteringConfiguration(document, current);
 		} else if (xmlCompare(current, "exportingProcess")) {
 			conf = new ExporterConfiguration(document, current);
+		} else if (xmlCompare(current, "pcapExporter")) {
+			conf = new PcapExporterConfiguration(document, current);
 		} else if (xmlCompare(current, "collectingProcess")) {
 			conf = new CollectorConfiguration(document, current);
 		} else if (xmlCompare(current, "dbWriter")) {
@@ -129,7 +134,7 @@ IpfixConfiguration::IpfixConfiguration(const std::string& configFile)
 			conf = new DbReaderConfiguration(document, current);
 #else
 			msg(MSG_ERROR, "IpfixConfiguration: Vermont was compiled without "
-				       "support for dbReader. Ignoring entry in conifg file!");
+				       "support for dbReader. Ignoring entry in config file!");
 #endif
 		}
 		if (conf) {
@@ -171,12 +176,13 @@ void IpfixConfiguration::connectSubsystems()
 	std::string TYPES[] = {
 		configTypes::observer,
 		configTypes::exporter,
+		configTypes::pcapExporter,
 		configTypes::dbwriter,
 		configTypes::dbreader,
 		configTypes::collector,
 		configTypes::metering,
 	};
-	for (unsigned t = 0; t != 6; ++t) {
+	for (unsigned t = 0; t != 7; ++t) {
 		for (SubsystemConfiguration::iterator i = subsystems.begin();
 		     i != subsystems.end(); ++i) {	
 			std::string id = i->first;
diff --git a/ipfix_configuration.h b/ipfix_configuration.h
index 5d382eb..d625c44 100644
--- a/ipfix_configuration.h
+++ b/ipfix_configuration.h
@@ -28,13 +28,14 @@ class ExporterConfiguration;
 
 namespace configTypes
 {
-	const std::string observer  = "observer";
-	const std::string exporter  = "exporter";
-	const std::string collector = "collector";
-	const std::string metering  = "metering";
-	const std::string dbwriter  = "dbwriter";
-	const std::string dbreader  = "dbreader";
-	const std::string main      = "main";
+	const std::string observer     = "observer";
+	const std::string exporter     = "exporter";
+	const std::string collector    = "collector";
+	const std::string metering     = "metering";
+	const std::string dbwriter     = "dbwriter";
+	const std::string dbreader     = "dbreader";
+	const std::string main         = "main";
+	const std::string pcapExporter = "pcapexporter";
 };
 
 
diff --git a/metering_configuration.cc b/metering_configuration.cc
index 89a3acd..20c89c4 100644
--- a/metering_configuration.cc
+++ b/metering_configuration.cc
@@ -6,12 +6,14 @@
 
 #include "metering_configuration.h"
 #include "exporter_configuration.h"
+#include "pcapexporter_configuration.h"
 #include "packetselection_configuration.h"
 #include "packetreporting_configuration.h"
 #include "flowmetering_configuration.h"
 #include "dbwriter_configuration.h"
 
 #include <sampler/Filter.h>
+#include <sampler/PcapExporterSink.h>
 #include <sampler/ExporterSink.h>
 #include <sampler/HookingFilter.h>
 #include <sampler/ExpressHookingFilter.h>
@@ -52,6 +54,11 @@ void MeteringConfiguration::setCaptureLength(int len)
 	captureLength = len;
 }
 
+void MeteringConfiguration::setDataLinkType(int type)
+{
+	dataLinkType = type;
+}
+
 void MeteringConfiguration::configure()
 {
 	msg(MSG_INFO, "MeteringConfiguration: Start reading meteringProcess");
@@ -98,6 +105,7 @@ void MeteringConfiguration::connect(Configuration* c)
 	// - an metering process (if the source does PacketSelection
 	//   and the destination does FlowMetering or PacketReporting
 	// - an dbWriter (if it does FlowMetering)
+	// - a pcapexporter process
 
 	ExporterConfiguration* exporter = dynamic_cast<ExporterConfiguration*>(c);
 	if (exporter) {
@@ -201,6 +209,14 @@ void MeteringConfiguration::connect(Configuration* c)
 		return;
 	}
 #endif
+	
+	PcapExporterConfiguration* pcapExporter = dynamic_cast<PcapExporterConfiguration*>(c);
+	if (pcapExporter) {
+		msg(MSG_DEBUG, "MeteringConfiguration: Adding pcapExporter to filter");
+		pcapExporter->getExporterSink()->setDataLinkType(dataLinkType);
+		getPacketSelectionConfiguration()->filter->setReceiver(pcapExporter->getExporterSink());
+		return;
+	}
 
 	THROWEXCEPTION("Cannot connect %s to metering process!", c->getId().c_str());
 }
diff --git a/metering_configuration.h b/metering_configuration.h
index 13893ca..7411080 100644
--- a/metering_configuration.h
+++ b/metering_configuration.h
@@ -33,6 +33,7 @@ public:
 
 	void setObservationDomainId(uint16_t id);
 	void setCaptureLength(int len);
+	void setDataLinkType(int type);
 
 	FlowMeteringConfiguration* getFlowMeteringConfiguration() { return flowMetering; }
 	FlowMeteringConfiguration* getExpressFlowMeteringConfiguration() { return expressflowMetering; }
@@ -46,6 +47,7 @@ private:
 
 	uint16_t observationDomainId;
 	int captureLength;
+	int dataLinkType;
 };
 
 #endif
diff --git a/observer_configuration.cc b/observer_configuration.cc
index 9b1500f..0e9ad8f 100644
--- a/observer_configuration.cc
+++ b/observer_configuration.cc
@@ -136,6 +136,7 @@ void ObserverConfiguration::connect(Configuration* c)
 		msg(MSG_DEBUG, "ObserverConfiguration: Connecting observer to metering process");
 		metering->setObservationDomainId(observationDomain);
 		metering->setCaptureLength(captureLength);
+		metering->setDataLinkType(observer->getDataLinkType());
 		PacketSelectionConfiguration* ps = metering->getPacketSelectionConfiguration();
 		observer->addReceiver(ps->getFilters());
 		return;
diff --git a/packetselection_configuration.cc b/packetselection_configuration.cc
index cd46a5b..fbfaa99 100644
--- a/packetselection_configuration.cc
+++ b/packetselection_configuration.cc
@@ -14,6 +14,8 @@
 #include <sampler/PacketSink.h>
 #include <sampler/stringFilter.h>
 #include <sampler/regExFilter.h>
+#include <sampler/ConnectionFilter.h>
+#include <sampler/StateConnectionFilter.h>
 
 
 PacketSelectionConfiguration::PacketSelectionConfiguration(xmlDocPtr document, xmlNodePtr startPoint)
@@ -156,6 +158,44 @@ void PacketSelectionConfiguration::configure()
 				j = j->next;
 			}
 			filter->addProcessor(new IPHeaderFilter(header, offset, size, comp, value));
+		} else if (tagMatches(i, "connectionFilter")) {
+#ifdef HAVE_CONNECTION_FILTER
+			xmlNodePtr j = i->xmlChildrenNode;
+			int bytes = 0;
+			int timeout = 0;
+			int size = 0;
+			int hashFunctions = 0;
+			msg(MSG_INFO, "PacketSelectionConfiguration: Creating connection filter");
+			while (NULL != j) {
+				if (tagMatches(j, "exportBytes")) {
+					bytes = atoi(getContent(j).c_str());
+				} else if (tagMatches(j, "timeout")) {
+					timeout = atoi(getContent(j).c_str());
+				} else if (tagMatches(j, "filterSize")) {
+					size = atoi(getContent(j).c_str());
+				} else if (tagMatches(j, "hashFunctions")) {
+					hashFunctions = atoi(getContent(j).c_str());
+				}
+				j = j->next;
+			}
+			filter->addProcessor(new ConnectionFilter(timeout, bytes, hashFunctions, size));
+#else
+			THROWEXCEPTION("Cannot configure ConnectionFilter because connection filter was disabled at compile time");
+#endif
+		} else if (tagMatches(i, "stateConnectionFilter")) {
+			xmlNodePtr j = i->xmlChildrenNode;
+			int bytes = 0;
+			int timeout = 0;
+			msg(MSG_INFO, "PacketSelectionConfiguration: Creating connection filter");
+			while (NULL != j) {
+				if (tagMatches(j, "exportBytes")) {
+					bytes = atoi(getContent(j).c_str());
+				} else if (tagMatches(j, "timeout")) {
+					timeout = atoi(getContent(j).c_str());
+				}
+				j = j->next;
+			}
+			filter->addProcessor(new StateConnectionFilter(timeout, bytes));
 		}
 		i = i->next;
 	}
diff --git a/pcapexporter_configuration.cc b/pcapexporter_configuration.cc
new file mode 100644
index 0000000..787e042
--- /dev/null
+++ b/pcapexporter_configuration.cc
@@ -0,0 +1,59 @@
+#include "pcapexporter_configuration.h"
+#include "metering_configuration.h"
+
+#include <sampler/PcapExporterSink.h>
+#include <common/msg.h>
+
+PcapExporterConfiguration::PcapExporterConfiguration(xmlDocPtr document, xmlNodePtr startPoint)
+	: Configuration(document, startPoint), exporterSink(0)
+{
+	xmlChar* idString = xmlGetProp(startPoint, (const xmlChar*)"id");
+	if (NULL == idString) {
+		THROWEXCEPTION("Got pcapExporter without unique id!");
+	}
+	id = configTypes::pcapExporter + (const char*)idString;
+	xmlFree(idString);
+}
+
+PcapExporterConfiguration::~PcapExporterConfiguration()
+{
+	delete exporterSink;
+}
+
+void PcapExporterConfiguration::configure()
+{
+	msg(MSG_INFO, "PcapExporterConfiguration: Start reading pcapExportingProcess section");
+	xmlNodePtr i = start->xmlChildrenNode;
+	while (NULL != i) {
+		if (tagMatches(i, "fileName")) {
+			fileName = getContent(i);
+		}
+		i = i->next;
+	}
+	setUp();
+	msg(MSG_INFO, "PcapExporterConfiguration: Successfully parsed PcapExportingProcess section");
+}
+
+void PcapExporterConfiguration::setUp()
+{
+	exporterSink = new PcapExporterSink(fileName);
+}
+
+void PcapExporterConfiguration::connect(Configuration*)
+{
+	THROWEXCEPTION("PcapExporter is an end target and cannot be connected to something!");
+}
+
+void PcapExporterConfiguration::startSystem()
+{
+	if (!exporterSink)
+		THROWEXCEPTION("Cannot start exporterSink, exporterSink does not exist!");
+	exporterSink->runSink();
+}
+
+void PcapExporterConfiguration::stopSystem()
+{
+	if (exporterSink)
+		exporterSink->terminateSink();
+}
+
diff --git a/pcapexporter_configuration.h b/pcapexporter_configuration.h
new file mode 100644
index 0000000..9674d26
--- /dev/null
+++ b/pcapexporter_configuration.h
@@ -0,0 +1,35 @@
+#ifndef _PCAP_EXPORTER_CONFIGURATION_H_
+#define _PCAP_EXPORTER_CONFIGURATION_H_
+
+
+#include "ipfix_configuration.h"
+
+
+
+#include <vector>
+
+
+class PcapExporterSink;
+
+
+class PcapExporterConfiguration : public Configuration {
+public:
+	PcapExporterConfiguration(xmlDocPtr document, xmlNodePtr startPoint);
+	~PcapExporterConfiguration();
+	
+	virtual void configure();
+	virtual void connect(Configuration*);
+	virtual void startSystem();
+	virtual void stopSystem();
+
+	PcapExporterSink* getExporterSink() const { return exporterSink; }
+	
+protected:
+	void setUp();
+
+private:
+	PcapExporterSink* exporterSink;
+	std::string fileName;
+};
+
+#endif
diff --git a/sampler/CMakeLists.txt b/sampler/CMakeLists.txt
index 1be8ba7..9fbf93c 100644
--- a/sampler/CMakeLists.txt
+++ b/sampler/CMakeLists.txt
@@ -32,5 +32,8 @@ ADD_LIBRARY(sampler
 	stringFilter.cpp
 	regExFilter.cpp
 	ExpressHookingFilter.cpp
+	ConnectionFilter.cpp
+	StateConnectionFilter.cpp
+	PcapExporterSink.cpp
 )
 
diff --git a/sampler/ConnectionFilter.cpp b/sampler/ConnectionFilter.cpp
new file mode 100644
index 0000000..6bf7944
--- /dev/null
+++ b/sampler/ConnectionFilter.cpp
@@ -0,0 +1,84 @@
+#ifdef HAVE_CONNECTION_FILTER
+
+#include "ConnectionFilter.h"
+
+ConnectionFilter::ConnectionFilter(unsigned Timeout, unsigned bytes, unsigned hashFunctions, unsigned filterSize)
+	:  hashParams(hashFunctions), synFilter(&hashParams, filterSize, false), exportFilter(&hashParams, filterSize, false),
+	  connectionFilter(&hashParams, filterSize, false), timeout(Timeout), exportBytes(bytes)
+{
+}
+
+ConnectionFilter::ConnectionFilter(unsigned Timeout, unsigned bytes, unsigned hashFunctions, unsigned filterSize, unsigned seed)
+	: hashParams(hashFunctions, seed), synFilter(&hashParams, filterSize, false), exportFilter(&hashParams, filterSize, false),
+	connectionFilter(&hashParams, filterSize, false), timeout(Timeout), exportBytes(bytes)
+{
+}
+
+bool ConnectionFilter::processPacket(const Packet* p)
+{
+	unsigned flagsOffset = p->transportHeaderOffset + 13;
+	static const uint8_t SYN = 0x02;
+	static const uint8_t FIN = 0x01;
+	static const uint8_t RST = 0x04;
+	static unsigned long tmp;
+	QuintupleKey key(p);
+	unsigned payloadLen = p->data_length - p->payloadOffset;
+
+	if (p->ipProtocolType != Packet::TCP) {
+		DPRINTF("Got a non-TCP packet. Protocol-type is %i", p->ipProtocolType);
+		return false;
+	}
+
+	if (*((uint8_t*)p->data + flagsOffset) & SYN) {
+		DPRINTF("ConnectionFilter: Got SYN packet");
+		synFilter.set(key.data, key.len, (agetime_t)p->timestamp.tv_sec);
+		DPRINTF("ConnectionFilter: synFilter saved time %u", synFilter.get(key.data, key.len));
+		// do not export SYN packet, or should we?
+		return false;
+	} else if (*((uint8_t*)p->data + flagsOffset) & RST || *((uint8_t*)p->data + flagsOffset) & FIN) {
+		
+		DPRINTF("ConnectionFilter: Got %s packet", *((uint8_t*)p->data + flagsOffset) & RST?"RST":"FIN");
+	
+		exportFilter.set(key.data, key.len, -exportFilter.get(key.data, key.len));
+		connectionFilter.set(key.data, key.len, p->timestamp.tv_sec);
+		DPRINTF("ConnectionFilter: connectionFilter saved time %u", connectionFilter.get(key.data, key.len));
+
+		// do not export FIN/RST packets, or should we?
+		return false;
+	} else {
+		DPRINTF("ConnectionFilter: Got a normal packet");
+		if ((tmp = exportFilter.get(key.data, key.len)) > 0) {
+			DPRINTF("ConnectionFilter: Connection known, exporting packet");
+			static unsigned diffVal;
+			if (tmp > payloadLen)
+				diffVal = -payloadLen;
+			else
+				diffVal = -tmp;
+			exportFilter.set(key.data, key.len, diffVal);
+			if (exportFilter.get(key.data, key.len) <= 0) {
+				connectionFilter.set(key.data, key.len, p->timestamp.tv_sec);
+			}
+			DPRINTF("ConnectionFilter: We have to export %i bytes after exporting this packet", exportFilter.get(key.data, key.len));
+			return true;
+		} else {
+			if ((unsigned)(p->timestamp.tv_sec - synFilter.get(key.data, key.len)) < timeout &&
+			    synFilter.get(key.data, key.len) - connectionFilter.get(key.data, key.len) > 0) {
+			    	DPRINTF("ConnectionFilter: Found new connection, exporting packet");
+				if (payloadLen < exportBytes) {
+					exportFilter.set(key.data, key.len, exportBytes - payloadLen);
+				} else {
+					connectionFilter.set(key.data, key.len, p->timestamp.tv_sec);
+				}
+				DPRINTF("ConnectionFilter: We have to export %i bytes after exporting this packet", exportFilter.get(key.data, key.len));
+				return true;
+			}
+			DPRINTF("ConnectionFilter: Paket will not be exported");
+			return false;
+		}
+	}
+
+	msg(MSG_FATAL, "ConnectionFilter: SOMTHING IS SCRWED UP, YOU SHOULD NEVER SEE THIS MESSAGE!");
+	return false; // make compiler happy
+}
+
+#endif
diff --git a/sampler/ConnectionFilter.h b/sampler/ConnectionFilter.h
new file mode 100644
index 0000000..679ef07
--- /dev/null
+++ b/sampler/ConnectionFilter.h
@@ -0,0 +1,29 @@
+#ifdef HAVE_CONNECTION_FILTER
+
+#ifndef _CONNECTION_FILTER_H_
+#define _CONNECTION_FILTER_H_
+
+#include "PacketProcessor.h"
+
+#include <common/CountBloomFilter.h>
+#include <common/AgeBloomFilter.h>
+
+class ConnectionFilter : public PacketProcessor {
+public:
+	ConnectionFilter(unsigned timeout, unsigned bytes, unsigned hashFunctions, unsigned FilterSize);
+	ConnectionFilter(unsigned timeout, unsigned bytes, unsigned hashFunctions, unsigned FilterSize, unsigned seed);
+
+
+	virtual bool processPacket(const Packet* p);
+protected:
+	HashParams hashParams;
+	AgeBloomFilter synFilter;
+	CountBloomFilter exportFilter;
+	AgeBloomFilter connectionFilter;
+	unsigned timeout;
+	unsigned exportBytes;
+};
+
+#endif
+
+#endif
diff --git a/sampler/Observer.cpp b/sampler/Observer.cpp
index 0e7a174..0215745 100644
--- a/sampler/Observer.cpp
+++ b/sampler/Observer.cpp
@@ -199,7 +199,7 @@ void *Observer::observerThread(void *arg)
 
 			// initialize packet structure (init copies packet data)
 			p = obs->packetManager->getNewInstance();
-			p->init((char*)pcapData, packetHeader.caplen, packetHeader.ts);
+			p->init((char*)pcapData, packetHeader.caplen, packetHeader.ts, packetHeader.len);
 
 			obs->receivedBytes += packetHeader.caplen;
 
@@ -235,6 +235,11 @@ void *Observer::observerThread(void *arg)
 		struct timeval first = {0,0};
 		// differences
 		struct timeval wait_val, delta_now, delta_file, delta_to_be;
+		
+		// make compiler happy ...
+		delta_to_be.tv_sec = 0;
+		delta_to_be.tv_usec = 0;
+
 		struct timespec wait_spec;
 		bool firstPacket = true;
 		// let's wait one seconds until other modules are ready
@@ -295,8 +300,9 @@ void *Observer::observerThread(void *arg)
 			p->init((char*)pcapData, 
 				// in constrast to live capturing, the data length is not limited
 				// to any snap length when reading from a pcap file
-				((int)packetHeader.caplen < obs->capturelen) ? (int)packetHeader.caplen : obs->capturelen, 
-				packetHeader.ts);
+				(packetHeader.caplen < obs->capturelen) ? packetHeader.caplen : obs->capturelen, 
+				packetHeader.ts,
+				packetHeader.len);
 
 			obs->receivedBytes += packetHeader.caplen;
 
@@ -337,7 +343,6 @@ void *Observer::observerThread(void *arg)
  */
 bool Observer::prepare(const std::string& filter)
 {
-	int dataLink = 0;
 	struct in_addr i_netmask, i_network;
 
 	// we need to store the filter expression, because pcap needs
@@ -376,9 +381,8 @@ bool Observer::prepare(const std::string& filter)
 			goto out2;
 		}
 
-		dataLink = pcap_datalink(captureDevice);
 		// IP_HEADER_OFFSET is set by the configure script
-		switch (dataLink) {
+		switch (getDataLinkType()) {
 		case DLT_EN10MB:
 			if (IP_HEADER_OFFSET != 14 && IP_HEADER_OFFSET != 18) {
 				msg(MSG_FATAL, "IP_HEADER_OFFSET on an ethernet device has to be 14 or 18 Bytes. Please adjust that value via configure --with-ipheader-offset");
@@ -392,6 +396,11 @@ bool Observer::prepare(const std::string& filter)
 				goto out2;
 			}
 			break;
+		case DLT_LINUX_SLL:
+			if (IP_HEADER_OFFSET != 16) {
+				msg(MSG_FATAL, "IP_HEADER_OFFSET on linux cooked devices has to be 16 Bytes. Please adjust that value via configure --with-ipheader-offset");
+				goto out2;
+			}
 		default:
 			msg(MSG_ERROR, "You are using an unkown IP_HEADER_OFFSET and data link combination. This can make problems. Please check if you use the correct IP_HEADER_OFFSET for your data link, if you see strange IPFIX/PSAMP packets.");
 		}
@@ -509,6 +518,11 @@ int Observer::getCaptureLen()
 	return capturelen;
 }
 
+int Observer::getDataLinkType()
+{
+	return pcap_datalink(captureDevice);
+}
+
 void Observer::replaceOfflineTimestamps()
 {
 	replaceTimestampsFromFile = true;
diff --git a/sampler/Observer.h b/sampler/Observer.h
index a5ce10d..569404d 100644
--- a/sampler/Observer.h
+++ b/sampler/Observer.h
@@ -57,6 +57,7 @@ public:
 	bool prepare(const std::string& filter);
 	static void doLogging(void *arg);
 	virtual std::string getStatistics();
+	int getDataLinkType();
 
 
 protected:
@@ -78,7 +79,7 @@ protected:
 	char errorBuffer[PCAP_ERRBUF_SIZE];
 
 	// also called snaplen; only sniff this much bytes from each packet
-	int capturelen;
+	unsigned capturelen;
 
 	// wait this much ms until pcap_read() returns and get ALL packets received
 	int pcap_timeout;
diff --git a/sampler/Packet.h b/sampler/Packet.h
index d4dc724..ba4ea2d 100644
--- a/sampler/Packet.h
+++ b/sampler/Packet.h
@@ -115,6 +115,8 @@ public:
 
 	// The number of captured bytes
 	unsigned int data_length;
+	// The pcap packet length
+	unsigned int pcap_packet_length;
 
 	// when was the packet received?
 	struct timeval timestamp;
@@ -140,7 +142,7 @@ public:
 	{
 	}
 	
-	inline void init(char* packetData, int len, struct timeval time) 
+	inline void init(char* packetData, unsigned int len, struct timeval time, unsigned int packet_length) 
 	{
 		transportHeader = NULL;
 		payload = NULL;
@@ -151,6 +153,7 @@ public:
 		timestamp = time;
 		varlength_index = 0;
 		ipProtocolType = NONE;
+		pcap_packet_length = packet_length;
 
 		if (len > PCAP_MAX_CAPTURE_LENGTH) {
 			THROWEXCEPTION("received packet of size %d is bigger than maximum length (%d), "
diff --git a/sampler/PacketReceiver.cpp b/sampler/PacketReceiver.cpp
index f9a31fd..119c287 100644
--- a/sampler/PacketReceiver.cpp
+++ b/sampler/PacketReceiver.cpp
@@ -3,7 +3,7 @@
 #include <sstream>
 
 
-PacketReceiver::PacketReceiver(string ownerName)
+PacketReceiver::PacketReceiver(const std::string& ownerName)
         : name(ownerName)
 {
         queue.setOwner(ownerName);
@@ -21,7 +21,7 @@ PacketReceiver::~PacketReceiver()
  */
 std::string PacketReceiver::getStatistics()
 {
-        ostringstream oss;
+        std::ostringstream oss;
         oss << "PacketReceiver(" << name << "): current queue size: " << queue.getCount();
 
         return oss.str();
diff --git a/sampler/PacketReceiver.h b/sampler/PacketReceiver.h
index 97ec4e4..c977a18 100644
--- a/sampler/PacketReceiver.h
+++ b/sampler/PacketReceiver.h
@@ -18,17 +18,14 @@
 
 #include <string>
 
-using namespace std;
-
-
 class PacketReceiver : public StatisticsModule
 {
 private:
 	ConcurrentQueue<Packet*>  queue;
-	string					  name;
+	std::string name;
 
 public:
-	PacketReceiver(string ownerName);
+	PacketReceiver(const std::string& ownerName);
 
 	virtual ~PacketReceiver();
 
diff --git a/sampler/PcapExporterSink.cpp b/sampler/PcapExporterSink.cpp
new file mode 100644
index 0000000..45232eb
--- /dev/null
+++ b/sampler/PcapExporterSink.cpp
@@ -0,0 +1,49 @@
+#include "PcapExporterSink.h"
+
+PcapExporterSink::PcapExporterSink(const std::string& fileName)
+	: Sink("PcapExporterSink"), thread(PcapExporterSink::pcapExporterSink), fileName(fileName),
+	exitFlag(false)
+{
+}
+
+PcapExporterSink::~PcapExporterSink()
+{
+	msg(MSG_DEBUG, "PcapExporterSink: destructor called");
+}
+
+void PcapExporterSink::setDataLinkType(int type)
+{
+	char errbuf[PCAP_ERRBUF_SIZE];
+	dummy = pcap_open_dead(type, 65535);
+	if (!dummy) {
+		THROWEXCEPTION("Could not open dummy device: %s", errbuf);
+	}
+}
+
+void* PcapExporterSink::pcapExporterSink(void* data)
+{
+	PcapExporterSink* sink = static_cast<PcapExporterSink*>(data);
+	ConcurrentQueue<Packet*> *queue = sink->getQueue();
+	Packet* p = NULL;
+	bool result;
+	pcap_pkthdr packetHeader;
+	char errbuf[PCAP_ERRBUF_SIZE];
+
+	sink->dumper = pcap_dump_open(sink->dummy, sink->fileName.c_str());
+	if (!sink->dumper) {
+		THROWEXCEPTION("Could not open dump file: %s", errbuf);
+	}
+
+	while (!sink->exitFlag) {
+		result = queue->pop(&p);
+		if (result) {
+			packetHeader.ts = p->timestamp;
+			packetHeader.caplen = p->data_length;
+			packetHeader.len = p->pcap_packet_length;
+			pcap_dump((unsigned char*)sink->dumper, &packetHeader, p->data);
+		}
+		p->removeReference();
+	}
+
+	return NULL;
+}
diff --git a/sampler/PcapExporterSink.h b/sampler/PcapExporterSink.h
new file mode 100644
index 0000000..17498e8
--- /dev/null
+++ b/sampler/PcapExporterSink.h
@@ -0,0 +1,54 @@
+#ifndef _PCAP_EXPORTER_SINK_H_
+#define _PCAP_EXPORTER_SINK_H_
+
+#include "Sink.h"
+
+#include <common/msg.h>
+#include <common/Thread.h>
+
+#include <string>
+#include <pcap.h>
+
+class PcapExporterSink : public Sink
+{
+public:
+	PcapExporterSink(const std::string& file);
+	~PcapExporterSink();
+
+        void runSink()
+        {
+                msg(MSG_DEBUG, "PcapExporterSink: now starting ExporterSink thread");
+                thread.run(this);
+        };
+
+        bool terminateSink()
+        {
+                exitFlag = true;
+                msg(MSG_DEBUG, "PcapExporterSink: waiting for exporter thread");
+                thread.join();
+                msg(MSG_DEBUG, "PcapExporterSink: exporter thread joined");
+		if (dumper) {
+			if (-1 == pcap_dump_flush(dumper)) {
+				msg(MSG_FATAL, "PcapExporterSink: Could not flush dump file");
+			}
+			pcap_dump_close(dumper);
+		}
+
+		return true;
+        };
+
+	void setDataLinkType(int type);
+
+
+
+private:
+	static void* pcapExporterSink(void* data);
+
+	Thread thread;
+	std::string fileName;
+	bool exitFlag;
+	pcap_t* dummy;
+	pcap_dumper_t* dumper;
+};
+
+#endif
diff --git a/sampler/Sink.h b/sampler/Sink.h
index ea7cc9e..3db48a7 100644
--- a/sampler/Sink.h
+++ b/sampler/Sink.h
@@ -15,7 +15,7 @@
 class Sink : public PacketReceiver {
 
 public:
-	Sink(string ownerName) : PacketReceiver(ownerName)
+	Sink(const std::string& ownerName) : PacketReceiver(ownerName)
 	{
 	}
 
diff --git a/sampler/StateConnectionFilter.cpp b/sampler/StateConnectionFilter.cpp
new file mode 100644
index 0000000..3793c96
--- /dev/null
+++ b/sampler/StateConnectionFilter.cpp
@@ -0,0 +1,60 @@
+#include "StateConnectionFilter.h"
+
+#include <iostream>
+
+StateConnectionFilter::StateConnectionFilter(unsigned timeout, unsigned bytes)
+{
+	this->timeout = timeout;
+	this->exportBytes = bytes;
+}
+
+bool StateConnectionFilter::processPacket(const Packet* p)
+{
+	return processPacket(p, true);
+}
+
+bool StateConnectionFilter::processPacket(const Packet* p, bool connFilterResult)
+{
+	unsigned flagsOffset = p->transportHeaderOffset + 13;
+	static const uint8_t SYN = 0x02;
+	static const uint8_t FIN = 0x01;
+	static const uint8_t RST = 0x04;
+	unsigned payloadLen = p->data_length - p->payloadOffset;
+
+	if (p->ipProtocolType != Packet::TCP) {
+		DPRINTF("Got a non-TCP packet. Protocol-type is %i", p->ipProtocolType);
+		return false;
+	}
+
+	QuintupleKey key(p);
+
+	if (*((uint8_t*)p->data + flagsOffset) & SYN) {
+		DPRINTF("StateConnectionFilter: Got SYN packet");
+		exportList[key] = exportBytes;
+		return false;
+	} else if (*((uint8_t*)p->data + flagsOffset) & RST || *((uint8_t*)p->data + flagsOffset) & FIN) {
+		DPRINTF("StateConnectionFilter: Got %s packet", *((uint8_t*)p->data + flagsOffset) & RST?"RST":"FIN");
+		if (exportList.find(key) != exportList.end()) {
+			exportList.erase(exportList.find(key));
+		}
+		return false;
+	} else {
+		DPRINTF("StateConnectionFilter: Got a normal packet");
+		if (exportList.find(key) != exportList.end() && exportList.find(key)->second > 0) {
+			DPRINTF("StateConnectionFilter: Connection known, exporting packet");
+			exportList[key] -= payloadLen;
+			DPRINTF("StateConnectionFilter: We have to export %i bytes after exporting this packet", exportList[key]>0?exportList[key]:0);
+			return true;
+		} else {
+			if (exportList.find(key) != exportList.end()) {
+				exportList.erase(exportList.find(key));
+			}
+			return false;
+		}
+	}
+
+	msg(MSG_FATAL, "StateConnectionFilter: SOMTHING IS SCRWED UP, YOU SHOULD NEVER SEE THIS MESSAGE!");
+	return false; // make compiler happy
+}
+
+
diff --git a/sampler/StateConnectionFilter.h b/sampler/StateConnectionFilter.h
new file mode 100644
index 0000000..9ef64ad
--- /dev/null
+++ b/sampler/StateConnectionFilter.h
@@ -0,0 +1,24 @@
+#ifndef _STATE_CONNECITON_FILTER_H_
+#define _STATE_CONNECTION_FILTER_H_
+
+#include <sampler/PacketProcessor.h>
+#include <common/BloomFilter.h>
+
+#include <map>
+#include <ostream>
+
+class MemStatistics;
+
+class StateConnectionFilter : public PacketProcessor {
+public:
+	StateConnectionFilter(unsigned timeout, unsigned bytes);
+
+	bool processPacket(const Packet* p, bool connFilterResult);
+	virtual bool processPacket(const Packet* p);
+protected:
+	unsigned timeout;
+	unsigned exportBytes;
+	std::map<QuintupleKey, int> exportList;
+};
+
+#endif
diff --git a/test_build.sh b/test_build.sh
index b837b4e..9d9703e 100755
--- a/test_build.sh
+++ b/test_build.sh
@@ -6,6 +6,8 @@ then
 	job_numbers=$1
 fi
 
+rm -f CMakeCache.txt
+
 (mkdir -p build000 && cd build000 && cmake -DDEBUG=no  -DSUPPORT_MYSQL=no  -DUSE_PCAPMMAP=no  .. && make clean && make -j $job_numbers && cd ..) || exit 1
 (mkdir -p build001 && cd build001 && cmake -DDEBUG=no  -DSUPPORT_MYSQL=no  -DUSE_PCAPMMAP=yes .. && make clean && make -j $job_numbers && cd ..) || exit 1
 (mkdir -p build010 && cd build010 && cmake -DDEBUG=no  -DSUPPORT_MYSQL=yes -DUSE_PCAPMMAP=no  .. && make clean && make -j $job_numbers && cd ..) || exit 1
diff --git a/tests/AggregationPerfTest.cpp b/tests/AggregationPerfTest.cpp
index aad585f..c5d9316 100644
--- a/tests/AggregationPerfTest.cpp
+++ b/tests/AggregationPerfTest.cpp
@@ -62,7 +62,7 @@ void AggregationPerfTest::expressTest()
 	shutdown();
 }
 
-Rule::Field* AggregationPerfTest::createRuleField(const string& typeId)
+Rule::Field* AggregationPerfTest::createRuleField(const std::string& typeId)
 {
 	Rule::Field* ruleField = new Rule::Field();
 	ruleField->modifier = Rule::Field::KEEP;
@@ -153,7 +153,7 @@ void AggregationPerfTest::start(unsigned int numpackets)
 	ConcurrentQueue<Packet*>* filterq = filter->getQueue();
 	for (unsigned int i=0; i<numpackets; i++) {
 		Packet* p = packetManager->getNewInstance();
-		p->init((char*)packetdata, packetdatalen, curtime);
+		p->init((char*)packetdata, packetdatalen, curtime, packetdatalen);
 		filterq->push(p);
 	}
 }
diff --git a/tests/AggregationPerfTest.h b/tests/AggregationPerfTest.h
index a7d94c8..344338e 100644
--- a/tests/AggregationPerfTest.h
+++ b/tests/AggregationPerfTest.h
@@ -32,7 +32,7 @@ class AggregationPerfTest
 		IpfixAggregator* ipfixAggregator;
 		InstanceManager<Packet>* packetManager;
 
-		Rule::Field* createRuleField(const string& typeId);
+		Rule::Field* createRuleField(const std::string& typeId);
 		Rules* createRules();
 
 		int numPackets;
diff --git a/tests/BloomFilterTest.cpp b/tests/BloomFilterTest.cpp
new file mode 100644
index 0000000..c655097
--- /dev/null
+++ b/tests/BloomFilterTest.cpp
@@ -0,0 +1,139 @@
+#ifdef HAVE_CONNECTION_FILTER
+
+#include "BloomFilterTest.h"
+#include <common/BloomFilter.h>
+#include <common/AgeBloomFilter.h>
+#include <common/CountBloomFilter.h>
+#include <common/msg.h>
+
+#include <iostream>
+
+#include <ctime>
+
+static void startTests();
+
+static QuintupleKey key1;
+static QuintupleKey key2;
+
+BloomFilterTestSuite::BloomFilterTestSuite()
+        : test_suite("BloomFilterTest")
+{
+        add(BOOST_TEST_CASE(&startTests));
+}
+
+static void setupGlobalKey()
+{
+	key1.getQuintuple()->srcIp = 12345678; //
+	key1.getQuintuple()->dstIp = 87654321; //
+	key1.getQuintuple()->proto = 17;
+	key1.getQuintuple()->srcPort = 11234;
+	key1.getQuintuple()->dstPort = 80;
+
+	key2.getQuintuple()->srcIp = 134567832;
+	key2.getQuintuple()->dstIp = 123543688;
+	key2.getQuintuple()->proto = 17;
+	key1.getQuintuple()->srcPort = 12323;
+	key1.getQuintuple()->dstPort = 32432;
+
+}
+
+static void testBloomFilter()
+{
+	HashParams hashParams(10);
+	BloomFilter* bf = new BloomFilter(&hashParams, 1000);
+
+	BOOST_REQUIRE(bf->get(key1.data, key1.len) == false);
+	BOOST_REQUIRE(bf->get(key1.data, key1.len) == false);
+
+	bf->set(key1.data, key1.len);
+	BOOST_REQUIRE(bf->get(key1.data, key1.len) == true);
+	
+	BOOST_REQUIRE(bf->get(key2.data, key2.len) == false);
+	bf->set(key2.data, key2.len);
+	BOOST_REQUIRE(bf->get(key2.data, key2.len) == true);
+	BOOST_REQUIRE(bf->get(key1.data, key1.len) == true);
+
+	delete bf;
+}
+
+static void testCountBloomFilter()
+{
+	HashParams hashParams(10);
+	CountBloomFilter* bf = new CountBloomFilter(&hashParams, 1000);
+
+	std::cout << "bf(key1) == " << bf->get(key1.data, key1.len) << std::endl;
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == 0);
+        std::cout << "bf(key2) == " << bf->get(key2.data, key2.len) << std::endl;
+        BOOST_REQUIRE(bf->get(key2.data, key2.len) == 0);
+	
+        bf->set(key1.data, key1.len, 100);
+	std::cout << "bf(key1) == " << bf->get(key1.data, key1.len) << std::endl;
+	std::cout << "bf(key2) == " << bf->get(key2.data, key2.len) << std::endl;
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == 100);
+        BOOST_REQUIRE(bf->get(key2.data, key2.len) == 0);
+	
+        bf->set(key2.data, key2.len, 1000);
+	std::cout << "bf(key1) == " << bf->get(key1.data, key1.len) << std::endl;
+	std::cout << "bf(key2) == " << bf->get(key2.data, key2.len) << std::endl;
+        BOOST_REQUIRE(bf->get(key2.data, key2.len) == 1000);
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == 100);
+
+	bf->set(key1.data, key1.len, 100);
+	std::cout << "bf(key1) == " << bf->get(key1.data, key1.len) << std::endl;
+	BOOST_REQUIRE(bf->get(key1.data, key1.len) == 200);
+
+	bf->set(key1.data, key1.len, -200);
+	std::cout << "bf(key1) == " << bf->get(key1.data, key1.len) << std::endl;
+	BOOST_REQUIRE(bf->get(key1.data, key1.len) == 0);	
+
+	bf->set(key2.data, key1.len, -1000);
+	std::cout << "bf(key2) == " << bf->get(key2.data, key2.len) << std::endl;
+	BOOST_REQUIRE(bf->get(key2.data, key2.len) == 0);
+
+	delete bf;
+}
+
+static void testAgeBloomFilter()
+{
+	HashParams hashParams(10);
+	AgeBloomFilter* bf = new AgeBloomFilter(&hashParams, 1000);
+
+	time_t now = time(NULL);
+	time_t later = now + 10;
+
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == 0);
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == 0);
+
+        bf->set(key1.data, key1.len, now);
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == now);
+        BOOST_REQUIRE(bf->get(key2.data, key2.len) == 0);
+
+        bf->set(key2.data, key2.len, later);
+        BOOST_REQUIRE(bf->get(key2.data, key2.len) == later);
+        BOOST_REQUIRE(bf->get(key1.data, key1.len) == now);	
+
+	BOOST_REQUIRE(bf->get(key2.data, key2.len) > bf->get(key1.data, key1.len));
+
+	delete bf;
+}
+
+static void startTests()
+{
+	std::cout << "Running tests on BloomFilter classes" << std::endl;
+	msg_init();
+	msg_setlevel(100);
+	setupGlobalKey();
+
+	std::cout << "Testing BloomFilter..." << std::endl;
+	testBloomFilter();
+	
+	std::cout << "Testing AgeBloomFilter..." << std::endl;
+	testAgeBloomFilter();
+
+	std::cout << "Testing CountBloomFilter..." << std::endl;
+	testCountBloomFilter();
+
+	std::cout << "All tests on all BloomFilter classes passed" << std::endl;
+}
+
+#endif
diff --git a/tests/BloomFilterTest.h b/tests/BloomFilterTest.h
new file mode 100644
index 0000000..af12cf6
--- /dev/null
+++ b/tests/BloomFilterTest.h
@@ -0,0 +1,20 @@
+#ifdef HAVE_CONNECTION_FILTER
+
+#ifndef _BLOOMFILTER_TEST_H_
+#define _BLOOMFILTER_TEST_H_
+
+#include <boost/test/unit_test.hpp>
+
+using boost::unit_test::test_suite;
+
+
+class BloomFilterTestSuite : public test_suite
+{
+	public:
+		BloomFilterTestSuite();
+};
+
+
+#endif
+
+#endif
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index f3cd904..8e73d4e 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -21,6 +21,8 @@ ADD_EXECUTABLE(vermonttest
 	test_concentrator.cpp
 	AggregationPerfTest.cpp
 	VermontTest.cpp
+	BloomFilterTest.cpp
+	ConnectionFilterTest.cpp
 )
 
 TARGET_LINK_LIBRARIES(vermonttest
@@ -42,4 +44,8 @@ IF (MYSQL_FOUND)
 	)
 ENDIF (MYSQL_FOUND)
 
-
+IF (CONNECTION_FILTER)
+	TARGET_LINK_LIBRARIES(vermonttest
+		${GSL_LIBRARIES}
+	)
+ENDIF (CONNECTION_FILTER)
diff --git a/tests/ConnectionFilterTest.cpp b/tests/ConnectionFilterTest.cpp
new file mode 100644
index 0000000..3b42fa2
--- /dev/null
+++ b/tests/ConnectionFilterTest.cpp
@@ -0,0 +1,110 @@
+#ifdef HAVE_CONNECTION_FILTER
+
+#include "ConnectionFilterTest.h"
+
+#include <sampler/ConnectionFilter.h>
+#include <sampler/Packet.h>
+#include <common/InstanceManager.h>
+#include <common/msg.h>
+
+#include <pcap.h>
+#include <iostream>
+
+static void startTests();
+static pcap_t *captureDevice;
+static char errorBuffer[PCAP_ERRBUF_SIZE]; 
+
+
+ConnectionFilterTestSuite::ConnectionFilterTestSuite()
+        : test_suite("BloomFilterTest")
+{
+        add(BOOST_TEST_CASE(&startTests));
+}
+
+static Packet* getNextPacket(pcap_t* dev)
+{
+	static InstanceManager<Packet> packetManager;
+	const unsigned char* pcapData;
+	struct pcap_pkthdr packetHeader;
+	Packet* p; 
+
+	pcapData = pcap_next(dev, &packetHeader);
+	p = packetManager.getNewInstance();
+	p->init((char*)pcapData, packetHeader.caplen, packetHeader.ts, packetHeader.len);
+	
+	return p;
+}
+
+static void startTests()
+{
+	std::cout << "running tests on ConnectionFilter" << std::endl;
+	msg_init();
+	msg_setlevel(100);
+	captureDevice = pcap_open_offline("data/connectionfiltertest.pcap", errorBuffer);
+	if (!captureDevice) {
+		BOOST_ERROR(errorBuffer);
+	}
+
+	Packet* p;
+	ConnectionFilter connFilter(5, 200, 10, 1000);
+	
+	// first packet is a udp packet
+	p = getNextPacket(captureDevice);
+	BOOST_REQUIRE(connFilter.processPacket(p) == false);
+
+	// process six packets that come from a connection that did not have any syn packet
+	p = getNextPacket(captureDevice);
+	BOOST_REQUIRE(connFilter.processPacket(p) == false);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false);
+
+	// process a valid short 5 packets connection
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // SYN
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // SYN
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == true);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // FIN
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // RST
+	
+	// process a valid connection
+	// ignore the first syn packet
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // SYN
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // SYN
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == true);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == true);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == true);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // passed export limit
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false);
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // FIN
+        p = getNextPacket(captureDevice);
+        BOOST_REQUIRE(connFilter.processPacket(p) == false); // FIN
+	p = getNextPacket(captureDevice);
+	BOOST_REQUIRE(connFilter.processPacket(p) == false); // ACK
+ 
+
+	pcap_close(captureDevice);
+
+	std::cout << "All tests on ConnectionFilter passed" << std::endl;
+}
+
+#endif
diff --git a/tests/ConnectionFilterTest.h b/tests/ConnectionFilterTest.h
new file mode 100644
index 0000000..e1d834d
--- /dev/null
+++ b/tests/ConnectionFilterTest.h
@@ -0,0 +1,19 @@
+#ifdef HAVE_CONNECTION_FILTER
+
+#ifndef _CONNECTION_FILTER_TEST_H_
+#define _CONNECTION_FILTER_TEST_H_
+
+#include <boost/test/unit_test.hpp>
+
+using boost::unit_test::test_suite;
+
+
+class ConnectionFilterTestSuite : public test_suite
+{
+	public:
+		ConnectionFilterTestSuite();
+};
+
+#endif
+
+#endif
diff --git a/tests/Makefile.am b/tests/Makefile.am
deleted file mode 100644
index e0c7245..0000000
--- a/tests/Makefile.am
+++ /dev/null
@@ -1,32 +0,0 @@
-bin_PROGRAMS=vermonttest
-
-vermonttest_SOURCES=test_concentrator.cpp common/msg.h AggregationPerfTest.cpp AggregationPerfTest.h common/Time.h VermontTest.cpp VermontTest.h
-
-AM_CFLAGS=-I$(top_srcdir) `xml2-config --cflags` $(MYSQL_CFLAGS) -Wall -Werror
-AM_CXXFLAGS=$(AM_CFLAGS)
-AM_LDFLAGS= \
-	-L$(top_builddir)/concentrator \
-	-L$(top_builddir)/sampler \
-	-L$(top_builddir)/ipfixlolib \
-	`xml2-config --libs` \
-	$(MYSQL_LDFLAGS) \
-	$(BOOST_LDFLAGS) \
-	$(BOOST_FILESYSTEM_LIB) \
-	$(BOOST_REGEX_LIB) \
-	$(BOOST_UNIT_TEST_FRAMEWORK_LIB)
-
-
-vermonttest_LDADD= \
-	$(top_builddir)/concentrator/libconcentrator.a \
-	$(top_builddir)/sampler/libsampler.a \
-	$(top_builddir)/ipfixlolib/libipfixlo.a \
-	$(top_builddir)/common/libcommon.a \
-	-lpthread \
-	$(LPCAP)
-
-vermonttest_DEPENDENCIES= \
-	$(top_builddir)/concentrator/libconcentrator.a \
-	$(top_builddir)/sampler/libsampler.a \
-	$(top_builddir)/ipfixlolib/libipfixlo.a \
-	$(top_builddir)/common/libcommon.a 
-
diff --git a/tests/VermontTest.cpp b/tests/VermontTest.cpp
index 03bf64c..8fd9602 100644
--- a/tests/VermontTest.cpp
+++ b/tests/VermontTest.cpp
@@ -4,12 +4,13 @@
 
 #include "VermontTest.h"
 #include "AggregationPerfTest.h"
+#include "BloomFilterTest.h"
+#include "ConnectionFilterTest.h"
 #include "test_concentrator.h"
 
 using boost::unit_test::test_suite;
 
 
-
 test_suite* init_unit_test_suite(int argc, char* argv[])
 {
 	test_suite* test(BOOST_TEST_SUITE("Vermont Testsuite"));
@@ -18,8 +19,12 @@ test_suite* init_unit_test_suite(int argc, char* argv[])
 
 	if (argc>1 && strcmp(argv[1], "-perf")==0) perftest = true;
 
-	test->add(new AggregationPerfTestSuite(!perftest));
-	test->add(new ConcentratorTestSuite());
+	//test->add(new AggregationPerfTestSuite(!perftest));
+	//test->add(new ConcentratorTestSuite());
+#ifdef HAVE_CONNECTION_FILTER
+	test->add(new BloomFilterTestSuite());
+	test->add(new ConnectionFilterTestSuite());
+#endif
 
 	return test;
 }
diff --git a/tests/data/connectionfiltertest.pcap b/tests/data/connectionfiltertest.pcap
new file mode 100644
index 0000000000000000000000000000000000000000..448b229ccf351f8c439fb216078bdce79df63316
GIT binary patch
literal 1907
zcmaLXUr19?90%~<J)5?ze_O~<Fw+o(Ub60jM4_RMIZM(aGZw_$#+=jC%@q1W6Y8O|
zKQF}|g6uJ&hxTBg^|F~E2wEsn4-p~u;6sWJQG)CDJKi&nL+^nDce;Jf_xJnVd*{>1
zmkf*`Rwe_W;D@_6>fOI~?S><GO<P_jZ~+uI6ahGGj<MVL1KO77_#M-qQ}1uqyYn~l
zpdLSl<e$;M!DWR@OJz=Q=C?+~FGsEE=IV_7MfK~+Pw_&gY&0RKm@G<a-L0fnh5|k^
z?P%7bL0eutjl<DQrYS1Sbn~c{Jai}HRtE13vbpX~q1Z5N;3>m0hu1u%cj+)s*&#$N
z<zzPHSxro1SWFu5k$DA8os=y+WuGvml*^Pd*&&(erm2@JW<Bv77kjrAZ4BPz2~qJY
zns`)SJkQ0R9Pu-DayedrBHkW+I_XVuEah^)>Q<l;J4ahyd^R<qMiLjY;>=Wqwe3#l
zLHgzhix*CJ12Cg=kCB-ytYz;-`nzH<<dghSR|X)tW?W>gu+B`il^N$zFBv%yhs$5K
z%A^bVE6O`hDs-A>9uTsYxdinEoRm`cZ?g*|(te9s*E{uQA3DRuYuk!8rqq5H;=ZU5
zizB*X6Bqk)#6vt|0gB(Yir%L#_m3xLok^+pNHiZ4;-aUrYHf6NKul`xgVfLRR7OmS
z<lnk7<V>ovYSFW{I@E{yAWiBSzEZOH0LlzEO^W0-ly{z#dY@-LnUi@C^(am1miA1M
zNV^?H<Ig<sV|d!~;`i$X^#K>{+lusl@@Rc6pH7e%@Wo1FJzZV>!BV;mMq_x@>+?r~
z{iU&B_^O4DEW;yrE~4%s9r@QGj?~%{#r7oV#n5Y)kDN|~P*jWdLjcp$mKQIgRUPI<
z6bMlpWrDYbioj=;9%EKMYhzeQP7&7!gW;Si=u8iusZ+%xUWF>ol&Tn~DuPTtbJi9{
zaVTlj5mnw7^9dnpRgCj0@S*XK331L#m5`X!73&`JjK_s2Wt@oP3$wjT7U#SqzSk91
VUbQ(pJ1ye`W{lAd*dEy`{s+$Xl;{8e

literal 0
HcmV?d00001

diff --git a/tests/test_concentrator.cpp b/tests/test_concentrator.cpp
index bd10139..b365305 100644
--- a/tests/test_concentrator.cpp
+++ b/tests/test_concentrator.cpp
@@ -344,7 +344,7 @@ void test_parser_stability() {
 }
 
 
-void start_test()
+static void start_test()
 {
 	// set Vermont messaging subsystem's debug level
 	//msg_setlevel(MSG_DEFAULT+99);
diff --git a/tools/Makefile.am b/tools/Makefile.am
deleted file mode 100644
index a604e02..0000000
--- a/tools/Makefile.am
+++ /dev/null
@@ -1,9 +0,0 @@
-bin_PROGRAMS=testCollector
-
-testCollector_SOURCES=testCollector.cpp ../common/msg.h
-
-AM_CFLAGS=-I$(top_srcdir) -I$(top_srcdir)/concentrator
-AM_CXXFLAGS=$(AM_CFLAGS)
-AM_LDFLAGS=-L$(top_builddir)/concentrator -L$(top_builddir)/ipfixlolib -lpthread -lpcap
-testCollector_LDADD=$(top_builddir)/concentrator/libconcentrator.a $(top_builddir)/ipfixlolib/libipfixlo.a $(top_builddir)/common/libcommon.a
-testCollector_DEPENDENCIES=$(top_builddir)/concentrator/libconcentrator.a $(top_builddir)/ipfixlolib/libipfixlo.a $(top_builddir)/common/libcommon.a