implemented support for multiple proxies

Makefile

@@ -4,7 +4,7 @@
 #SUBDIR+=event
 #SUBDIR+=http
 #SUBDIR+=io
-SUBDIR+=programs
+SUBDIR+=proxy
 #SUBDIR+=ssh
 #SUBDIR+=xcodec
 #SUBDIR+=zlib

build-release

@@ -1,4 +1,4 @@
-cd programs/wanproxy
+cd proxy
 mkdir -p bin
 make NDEBUG=1
 

common/program.mk

@@ -76,7 +76,7 @@ endif
 CFLAGS+=-Wno-system-headers
 CFLAGS+=-Wno-unused-parameter
 CFLAGS+=-Wno-switch
-CFLAGS+=-Wno-uninitialized
+#CFLAGS+=-Wno-uninitialized
 CFLAGS+=-Wpointer-arith -Wreturn-type -Wcast-qual -Wwrite-strings -Wshadow -Wcast-align -Wchar-subscripts -Wreorder
 #CFLAGS+=-Winline
 CXXFLAGS+=-Wno-deprecated

programs/Makefile

@@ -1,3 +0,0 @@
-SUBDIR+=wanproxy
-
-include ../common/subdir.mk

proxy/Makefile

@@ -1,8 +1,5 @@
 PROGRAM=wanproxy
 
-SRCS+=	proxy_connector.cc
-SRCS+=	proxy_listener.cc
-
 SRCS+=	wanproxy.cc
 SRCS+=	wanproxy_config.cc
 SRCS+=	wanproxy_config_class_codec.cc
@@ -13,7 +10,10 @@ SRCS+=	wanproxy_config_type_codec.cc
 SRCS+=	wanproxy_config_type_compressor.cc
 SRCS+=	wanproxy_config_type_proxy_type.cc
 SRCS+=	wanproxy_config_type_proxy_role.cc
+SRCS+=	proxy_listener.cc
+SRCS+=	proxy_connector.cc
 
-TOPDIR=../..
+TOPDIR=..
 USE_LIBS=common common/thread common/time common/uuid config crypto event http io io/net io/socket ssh xcodec xcodec/cache/coss zlib
 include ${TOPDIR}/common/program.mk
+

proxy/proxy_connector.cc

@@ -154,8 +154,8 @@ bool ProxyConnector::build_chains (WANProxyCodec* cdc1, WANProxyCodec* cdc2, Soc
 		if (cdc1->xcache_) 
       {
 			EncodeFilter* enc; DecodeFilter* dec;
-			request_chain_.append ((dec = new DecodeFilter ("/wanproxy/" + cdc1->name_ + "/dec", cdc1->xcache_)));
-			response_chain_.prepend ((enc = new EncodeFilter ("/wanproxy/" + cdc1->name_ + "/enc", cdc1->xcache_, 1)));
+			request_chain_.append ((dec = new DecodeFilter ("/wanproxy/" + cdc1->name_ + "/dec", cdc1)));
+			response_chain_.prepend ((enc = new EncodeFilter ("/wanproxy/" + cdc1->name_ + "/enc", cdc1, 1)));
          dec->set_upstream (enc);
 		}
 
@@ -177,8 +177,8 @@ bool ProxyConnector::build_chains (WANProxyCodec* cdc1, WANProxyCodec* cdc2, Soc
 		if (cdc2->xcache_) 
       {
 			EncodeFilter* enc; DecodeFilter* dec;
-			request_chain_.append ((enc = new EncodeFilter ("/wanproxy/" + cdc2->name_ + "/enc", cdc2->xcache_)));
-			response_chain_.prepend ((dec = new DecodeFilter ("/wanproxy/" + cdc2->name_ + "/dec", cdc2->xcache_)));
+			request_chain_.append ((enc = new EncodeFilter ("/wanproxy/" + cdc2->name_ + "/enc", cdc2)));
+			response_chain_.prepend ((dec = new DecodeFilter ("/wanproxy/" + cdc2->name_ + "/dec", cdc2)));
          dec->set_upstream (enc);
 		}
 

proxy/wanproxy.cc

@@ -38,7 +38,7 @@
 //                                                                            //
 ////////////////////////////////////////////////////////////////////////////////
 
-#define PROGRAM_VERSION		"3.02"
+#define PROGRAM_VERSION		"3.0.5"
 
 WanProxyCore wanproxy;	
 
@@ -55,7 +55,7 @@ int main (int argc, char *argv[])
 
 	INFO("/wanproxy") << "WANProxy MT " << PROGRAM_VERSION;
 	INFO("/wanproxy") << "Copyright (c) 2008-2013 WANProxy.org";
-	INFO("/wanproxy") << "Copyright (c) 2013-2015 Bramfeld-Software";
+	INFO("/wanproxy") << "Copyright (c) 2013-2018 Bramfeld-Software";
 	INFO("/wanproxy") << "All rights reserved.";
 
 	while ((ch = getopt(argc, argv, "c:qv")) != -1) 
@@ -92,11 +92,10 @@ int main (int argc, char *argv[])
 	if (! wanproxy.configure (configfile)) 
 	{
 		ERROR("/wanproxy") << "Could not configure proxies.";
-		return (1);
+		wanproxy.terminate ();
+		return 1;
 	}
 	
-	wanproxy.launch_listener ();
-
 	event_system.run ();
 	
 	wanproxy.terminate ();
@@ -109,3 +108,4 @@ static void usage(void)
 	INFO("/wanproxy/usage") << "wanproxy [-q | -v] -c configfile";
 	exit(1);
 }
+

proxy/wanproxy.h

@@ -21,16 +21,8 @@
 #include "wanproxy_config_type_codec.h"
 #include "proxy_listener.h"
 
-
-struct WanProxyCore
+struct WanProxyInstance
 {
-private:
-	std::string config_file_;
-	Action* reload_action_;
-	ProxyListener* listener_;
-	std::map<UUID, XCodecCache*> caches_;
-	
-public:
 	std::string proxy_name_;
 	bool proxy_client_;
 	bool proxy_secure_;
@@ -40,17 +32,33 @@ public:
 	SocketAddressFamily remote_protocol_;
 	std::string remote_address_;
 	WANProxyCodec remote_codec_;
-	WANProxyConfigCache cache_type_;
-	std::string cache_path_;
-	size_t cache_size_;
-	UUID cache_uuid_;
+	ProxyListener* listener_;
+	
+	WanProxyInstance ()
+	{
+		proxy_client_ = proxy_secure_ = false; 
+		local_protocol_ = remote_protocol_ = SocketAddressFamilyIP;
+		listener_ = 0;
+	}
+	
+	~WanProxyInstance ()
+	{
+		delete listener_;
+	}
+};
 
+struct WanProxyCore
+{
+private:
+	std::string config_file_;
+	Action* reload_action_;
+	std::map<UUID, XCodecCache*> caches_;
+	std::map<std::string, WanProxyInstance> proxies_;
+
+public:
 	WanProxyCore ()
 	{
 		reload_action_ = 0;
-		listener_ = 0;
-		proxy_client_ = proxy_secure_ = false; local_protocol_ = remote_protocol_ = SocketAddressFamilyIP;
-		cache_type_ = WANProxyConfigCacheMemory; cache_size_ = 0;
 	}
 	
 	bool configure (const std::string& file)
@@ -60,34 +68,51 @@ public:
 		if (reload_action_)
 			reload_action_->cancel ();
 		reload_action_ = event_system.register_interest (EventInterestReload, callback (this, &WanProxyCore::reload));
-		return (! config_file_.empty () && config.configure (config_file_)); 
-	}
-	
-	void launch_listener ()
-	{
-		listener_ = new ProxyListener (proxy_name_, &local_codec_, &remote_codec_, local_protocol_, local_address_, 
-												 remote_protocol_, remote_address_, proxy_client_, proxy_secure_);
+		return config.read_file (config_file_); 
 	}
 	
 	void reload ()
 	{
-		if (configure (config_file_) && listener_) 
-			listener_->refresh (proxy_name_, &local_codec_, &remote_codec_, local_protocol_, local_address_, 
-									  remote_protocol_, remote_address_, proxy_client_, proxy_secure_);
+		if (configure (config_file_)) 
+			INFO("wanproxy/core") << "Reloaded proxy configuration.";
 		else
 			INFO("wanproxy/core") << "Could not reconfigure proxies.";
 	}	
 	
-	XCodecCache* add_cache (UUID uuid, size_t size)
+	void add_proxy (std::string& name, WanProxyInstance& data)
+	{
+	   WanProxyInstance& prx = proxies_[name];
+	   
+	   prx.proxy_name_ = data.proxy_name_;
+	   prx.proxy_client_ = data.proxy_client_;
+	   prx.proxy_secure_ = data.proxy_secure_;
+	   prx.local_protocol_ = data.local_protocol_;
+	   prx.local_address_ = data.local_address_;
+	   prx.local_codec_ = data.local_codec_;
+	   prx.remote_protocol_ = data.remote_protocol_;
+	   prx.remote_address_ = data.remote_address_;
+	   prx.remote_codec_ = data.remote_codec_;
+	   
+	   if (! prx.listener_)
+			prx.listener_ = new ProxyListener (prx.proxy_name_, &prx.local_codec_, &prx.remote_codec_, 
+														  prx.local_protocol_, prx.local_address_, prx.remote_protocol_, prx.remote_address_,
+														  prx.proxy_client_, prx.proxy_secure_);
+	   else
+			prx.listener_->refresh (prx.proxy_name_, &prx.local_codec_, &prx.remote_codec_, 
+											prx.local_protocol_, prx.local_address_, prx.remote_protocol_, prx.remote_address_, 
+											prx.proxy_client_, prx.proxy_secure_);
+	}
+	
+	XCodecCache* add_cache (WANProxyConfigCache type, std::string& path, size_t size, UUID& uuid)
 	{
 		XCodecCache* cache = 0;
-		switch (cache_type_)
+		switch (type)
 		{
 		case WANProxyConfigCacheMemory:
 			cache = new XCodecMemoryCache (uuid, size);
 			break;
 		case WANProxyConfigCacheCOSS: 
-			cache = new XCodecCacheCOSS (uuid, cache_path_, size);
+			cache = new XCodecCacheCOSS (uuid, path, size);
 			break;
 		}
 		ASSERT("/xcodec/cache", caches_.find(uuid) == caches_.end());
@@ -108,31 +133,37 @@ public:
 	{
 		if (reload_action_)
 			reload_action_->cancel (), reload_action_ = 0;
-		delete listener_; listener_ = 0;
-		print_stream_counts ();
+			
+		std::map<std::string, WanProxyInstance>::iterator prx;
+		for (prx = proxies_.begin(); prx != proxies_.end(); prx++)
+		   print_stream_counts (prx->second);
+		proxies_.clear ();
+		   
 		std::map<UUID, XCodecCache*>::iterator it;
 		for (it = caches_.begin(); it != caches_.end(); it++)
 			delete it->second;
 		caches_.clear ();
-		
 	}
 	
-	void print_stream_counts ()
+	void print_stream_counts (WanProxyInstance& prx)
 	{
-		if (local_codec_.counting_)
+		if (prx.local_codec_.counting_ || prx.remote_codec_.counting_)
+			INFO("wanproxy/core") << "Stream counts for proxy: " << prx.proxy_name_;
+		
+		if (prx.local_codec_.counting_)
 		{
-			INFO("wanproxy/core") << "Local codec request input bytes:    " << local_codec_.request_input_bytes_;
-			INFO("wanproxy/core") << "Local codec request output bytes:   " << local_codec_.request_output_bytes_;
-			INFO("wanproxy/core") << "Local codec response input bytes:   " << local_codec_.response_input_bytes_;
-			INFO("wanproxy/core") << "Local codec response output bytes:  " << local_codec_.response_output_bytes_;
+			INFO("wanproxy/core") << "Local codec request input bytes:    " << prx.local_codec_.request_input_bytes_;
+			INFO("wanproxy/core") << "Local codec request output bytes:   " << prx.local_codec_.request_output_bytes_;
+			INFO("wanproxy/core") << "Local codec response input bytes:   " << prx.local_codec_.response_input_bytes_;
+			INFO("wanproxy/core") << "Local codec response output bytes:  " << prx.local_codec_.response_output_bytes_;
 		}
 		
-		if (remote_codec_.counting_)
+		if (prx.remote_codec_.counting_)
 		{
-			INFO("wanproxy/core") << "Remote codec request input bytes:   " << remote_codec_.request_input_bytes_;
-			INFO("wanproxy/core") << "Remote codec request output bytes:  " << remote_codec_.request_output_bytes_;
-			INFO("wanproxy/core") << "Remote codec response input bytes:  " << remote_codec_.response_input_bytes_;
-			INFO("wanproxy/core") << "Remote codec response output bytes: " << remote_codec_.response_output_bytes_;
+			INFO("wanproxy/core") << "Remote codec request input bytes:   " << prx.remote_codec_.request_input_bytes_;
+			INFO("wanproxy/core") << "Remote codec request output bytes:  " << prx.remote_codec_.request_output_bytes_;
+			INFO("wanproxy/core") << "Remote codec response input bytes:  " << prx.remote_codec_.response_input_bytes_;
+			INFO("wanproxy/core") << "Remote codec response output bytes: " << prx.remote_codec_.response_output_bytes_;
 		}
 	}
 };

proxy/wanproxy_codec.h

@@ -26,6 +26,10 @@
 #ifndef	PROGRAMS_WANPROXY_WANPROXY_CODEC_H
 #define	PROGRAMS_WANPROXY_WANPROXY_CODEC_H
 
+#include "wanproxy_config_type_codec.h"
+#include "wanproxy_config_type_compressor.h"
+#include <xcodec/xcodec_cache.h>
+
 ////////////////////////////////////////////////////////////////////////////////
 //                                                                            //
 // File:           wanproxy_codec.h                                           //
@@ -36,10 +40,12 @@
 //                                                                            //
 ////////////////////////////////////////////////////////////////////////////////
 
-class XCodecCache;
-
 struct WANProxyCodec {
 	std::string name_;
+	WANProxyConfigCache cache_type_;
+	std::string cache_path_;
+	size_t cache_size_;
+	UUID cache_uuid_;
 	XCodecCache* xcache_;
 	bool compressor_;
 	char compressor_level_;
@@ -51,6 +57,8 @@ struct WANProxyCodec {
 
 	WANProxyCodec(void)
 	: name_(""),
+	  cache_type_(WANProxyConfigCacheMemory),
+	  cache_size_(0),
 	  xcache_(NULL),
 	  compressor_(false),
 	  compressor_level_(0),

proxy/wanproxy_config.cc

@@ -140,21 +140,24 @@ WANProxyConfig::parse_set(std::deque<std::string>& tokens)
 }
 
 bool
-WANProxyConfig::configure(const std::string& name)
+WANProxyConfig::read_file(const std::string& path)
 {
-	if (config_ != NULL) {
-		ERROR(log_) << "WANProxy already configured.";
-		return (false);
-	}
+	delete config_;
+	config_ = NULL;
 
 	INFO(log_) << "Configuring WANProxy.";
 
+	if (path.empty()) {
+		ERROR(log_) << "No file name specified";
+		return false;
+	}
+
 	std::fstream in;
-	in.open(name.c_str(), std::ios::in);
+	in.open(path.c_str(), std::ios::in);
 
 	if (!in.good()) {
-		ERROR(log_) << "Could not open file: " << name;
-		return (false);
+		ERROR(log_) << "Could not open file: " << path;
+		return false;
 	}
 
 	config_ = new Config();
@@ -187,17 +190,15 @@ WANProxyConfig::configure(const std::string& name)
 		}
 		if (tokens.empty())
 			continue;
-		ASSERT(log_, !tokens.empty());
 		if (!parse(tokens)) {
 			ERROR(log_) << "Error in configuration directive: " << line;
-
 			delete config_;
 			config_ = NULL;
-
-			return (false);
+			return false;
 		}
 		ASSERT(log_, tokens.empty());
 	}
 
-	return (true);
+	return true;
 }
+

proxy/wanproxy_config.h

@@ -53,7 +53,7 @@ private:
 	void parse_set(std::deque<std::string>&);
 
 public:
-	bool configure(const std::string&);
+	bool read_file(const std::string&);
 };
 
 #endif /* !PROGRAMS_WANPROXY_WANPROXY_CONFIG_H */

proxy/wanproxy_config_class_codec.cc

@@ -61,10 +61,7 @@ WANProxyConfigClassCodec::Instance::activate(const ConfigObject *co)
 		 */
 		if (cache_path_.empty())
 		{
-			if (wanproxy.cache_uuid_.is_valid ())
-				uuid = wanproxy.cache_uuid_;
-			else
-				uuid.generate();
+			uuid.generate();
 		}
 		else
 		{   
@@ -76,13 +73,13 @@ WANProxyConfigClassCodec::Instance::activate(const ConfigObject *co)
 			}
 		}
 
-		wanproxy.cache_type_ = cache_type_;
-		wanproxy.cache_path_ = cache_path_;
-		wanproxy.cache_size_ = local_size_;
-		wanproxy.cache_uuid_ = uuid;
+		codec_.cache_type_ = cache_type_;
+		codec_.cache_path_ = cache_path_;
+		codec_.cache_size_ = local_size_;
+		codec_.cache_uuid_ = uuid;
 
 		if (! (cache = wanproxy.find_cache (uuid)))
-			cache = wanproxy.add_cache (uuid, local_size_);
+			cache = wanproxy.add_cache (cache_type_, cache_path_, local_size_, uuid);
 		codec_.xcache_ = cache;
 		break;
 	case WANProxyConfigCodecNone:

proxy/wanproxy_config_class_proxy.cc

@@ -94,16 +94,18 @@ WANProxyConfigClassProxy::Instance::activate(const ConfigObject *co)
 	
 	if (role_ == WANProxyConfigProxyRoleUndefined && ! interface_codec && peer_codec)
 		role_ = WANProxyConfigProxyRoleClient;
-
-	wanproxy.proxy_name_ = co->name_;
-	wanproxy.proxy_client_ = (role_ == WANProxyConfigProxyRoleClient);
-	wanproxy.proxy_secure_ = (type_ == WANProxyConfigProxyTypeSSHSSH);
-	wanproxy.local_protocol_ = interface->family_;
-	wanproxy.local_address_ = '[' + interface->host_ + ']' + ':' + interface->port_;
-	wanproxy.local_codec_ = (interface_codec ? *interface_codec : WANProxyCodec ());
-	wanproxy.remote_protocol_ = peer->family_;
-	wanproxy.remote_address_ = '[' + peer->host_ + ']' + ':' + peer->port_;
-	wanproxy.remote_codec_ = (peer_codec ? *peer_codec : WANProxyCodec ());
-
+		
+	WanProxyInstance ins;
+	ins.proxy_name_ = co->name_;
+	ins.proxy_client_ = (role_ == WANProxyConfigProxyRoleClient);
+	ins.proxy_secure_ = (type_ == WANProxyConfigProxyTypeSSHSSH);
+	ins.local_protocol_ = interface->family_;
+	ins.local_address_ = '[' + interface->host_ + ']' + ':' + interface->port_;
+	ins.local_codec_ = (interface_codec ? *interface_codec : WANProxyCodec ());
+	ins.remote_protocol_ = peer->family_;
+	ins.remote_address_ = '[' + peer->host_ + ']' + ':' + peer->port_;
+	ins.remote_codec_ = (peer_codec ? *peer_codec : WANProxyCodec ());
+	wanproxy.add_proxy (ins.proxy_name_, ins);
+	
 	return (true);
 }

xcodec/cache/coss/xcodec_cache_coss.cc

@@ -91,13 +91,11 @@ XCodecCacheCOSS::~XCodecCacheCOSS()
 
 	delete[] directory_;
 
-	/*
 	INFO(log_) << "Stats: ";
 	INFO(log_) << "\tLookups=" << stats_.lookups;
 	INFO(log_) << "\tHits=" << (stats_.found_1 + stats_.found_2) << " (" << stats_.found_1 << " + " << stats_.found_2 << ")";
    if (stats_.lookups > 0)
       INFO(log_) << "\tHit ratio=" << ((stats_.found_1 + stats_.found_2) * 100) / stats_.lookups << "%";
-	*/
 
 	DEBUG(log_) << "Closing coss file: " << file_path_;
 	DEBUG(log_) << "Serial: " << serial_number_;
@@ -195,13 +193,15 @@ bool XCodecCacheCOSS::lookup (const uint64_t& hash, Buffer& buf)
 	int slot;
 
 	stats_.lookups++;
-	
+
+#ifdef USING_XCODEC_CACHE_RECENT_WINDOW
 	if ((data = find_recent (hash)))
 	{
 		buf.append (data, XCODEC_SEGMENT_LENGTH);
 		stats_.found_1++;
 		return true;
 	}
+#endif
 		
 	if (! (entry = cache_index_.lookup (hash)))
 		return false;
@@ -227,7 +227,9 @@ bool XCodecCacheCOSS::lookup (const uint64_t& hash, Buffer& buf)
 	stripe_[slot].header.flags[entry->position] |= 3;
 
 	data = stripe_[slot].segment_array[entry->position].bytes;
+#ifdef USING_XCODEC_CACHE_RECENT_WINDOW
 	remember (hash, data);
+#endif
 	buf.append (data, XCODEC_SEGMENT_LENGTH);
 	stats_.found_2++;
 	return true;
@@ -337,7 +339,9 @@ void XCodecCacheCOSS::detach_stripe (int slot)
 		{
 			if (stripe_[slot].header.flags[i] & 1)
 			{
+#ifdef USING_XCODEC_CACHE_RECENT_WINDOW
 				forget (stripe_[slot].header.hash_array[i]);
+#endif
 				stripe_[slot].header.flags[i] &= ~1;
 			}
 		}

xcodec/cache/coss/xcodec_cache_coss.h

@@ -79,7 +79,7 @@ using namespace std;
 #define CACHE_SIGNATURE				0xF150E964
 #define CACHE_VERSION				2
 #define STRIPE_SEGMENT_COUNT		512		// segments of XCODEC_SEGMENT_LENGTH per stripe (must fit into 16 bits)
-#define LOADED_STRIPE_COUNT		4			// number of stripes held in memory (must be greater than 1)
+#define LOADED_STRIPE_COUNT		16			// number of stripes held in memory (must be greater than 1)
 #define CACHE_BASIC_SIZE			1024		// MB
 
 #define CACHE_ALIGNEMENT			4096
@@ -89,6 +89,8 @@ using namespace std;
 #define HEADER_ALIGNED_SIZE		ROUND_UP(HEADER_ARRAY_SIZE + METADATA_SIZE, CACHE_ALIGNEMENT)
 #define METADATA_PADDING			(HEADER_ALIGNED_SIZE - HEADER_ARRAY_SIZE - METADATA_SIZE)
 
+#define USING_XCODEC_CACHE_RECENT_WINDOW
+
 struct COSSIndexEntry 
 {
 	uint64_t stripe_range : 48;

xcodec/xcodec_filter.cc

@@ -27,7 +27,6 @@
 #include <common/endian.h>
 #include <common/count_filter.h>
 #include <event/event_system.h>
-#include <programs/wanproxy/wanproxy.h>
 #include "xcodec_filter.h"
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -239,7 +238,7 @@ bool DecodeFilter::consume (Buffer& buf, int flg)
 				ERROR(log_) << "Got <HELLO> twice.";
 				return false;
 			}
-			else
+			else if (codec_)
 			{
 		      uint8_t len;
 		      if (pending_.length() < sizeof op + sizeof len)
@@ -266,7 +265,7 @@ bool DecodeFilter::consume (Buffer& buf, int flg)
 		      pending_.skip (sizeof mb);
 
 				if (! (decoder_cache_ = wanproxy.find_cache (uuid)))
-					decoder_cache_ = wanproxy.add_cache (uuid, mb);
+					decoder_cache_ = wanproxy.add_cache (codec_->cache_type_, codec_->cache_path_, mb, uuid);
 
 		      ASSERT(log_, decoder_ == NULL);
 				if (decoder_cache_)

xcodec/xcodec_filter.h

@@ -20,11 +20,13 @@
 #include <xcodec/xcodec_hash.h>
 #include <xcodec/xcodec_encoder.h>
 #include <xcodec/xcodec_decoder.h>
+#include <proxy/wanproxy.h>
 
 class EncodeFilter : public BufferedFilter
 {
 private:
-   XCodecCache* cache_;
+	WANProxyCodec* codec_;
+	XCodecCache* cache_;
 	XCodecEncoder* encoder_;
 	Action* wait_action_;
 	bool waiting_;
@@ -32,9 +34,10 @@ private:
 	bool eos_ack_;
    
 public:
-	EncodeFilter (const LogHandle& log, XCodecCache* cc, int flg = 0) : BufferedFilter (log) 
+	EncodeFilter (const LogHandle& log, WANProxyCodec* cdc, int flg = 0) : BufferedFilter (log) 
 	{ 
-		cache_ = cc; encoder_ = 0; wait_action_ = 0; waiting_ = (flg & 1); sent_eos_ = eos_ack_ = false;
+		codec_ = cdc; cache_ = (cdc ? cdc->xcache_ : 0); encoder_ = 0; 
+		wait_action_ = 0; waiting_ = (flg & 1); sent_eos_ = eos_ack_ = false;
 	}
 	
 	virtual ~EncodeFilter ()  
@@ -55,7 +58,8 @@ private:
 class DecodeFilter : public LogisticFilter
 {
 private:
-   XCodecCache* encoder_cache_;
+	WANProxyCodec* codec_;
+	XCodecCache* encoder_cache_;
 	XCodecDecoder* decoder_;
 	XCodecCache* decoder_cache_;
 	std::set<uint64_t> unknown_hashes_;
@@ -66,9 +70,9 @@ private:
 	bool upflushed_;
    
 public:
-	DecodeFilter (const LogHandle& log, XCodecCache* cc) : LogisticFilter (log) 
+	DecodeFilter (const LogHandle& log, WANProxyCodec* cdc) : LogisticFilter (log) 
    { 
-      encoder_cache_ = cc; decoder_ = 0; decoder_cache_ = 0;   
+      codec_ = cdc; encoder_cache_ = (cdc ? cdc->xcache_ : 0); decoder_ = 0; decoder_cache_ = 0;   
       received_eos_ = sent_eos_ack_ = received_eos_ack_ = upflushed_ = false; 
    }