

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.0.0.170915_Beta4 (15 September 2017)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2017
//_/_/  
//_/_/  in collaboration with 
//_/_/  
//_/_/  Icelandic Institute for Intelligent Machines ses
//_/_/    Menntavegur 1
//_/_/    101 Reykjavik
//_/_/    ICELAND
//_/_/    http://www.iiim.is/
//_/_/
//_/_/  and is released under the following
//_/_/  
//_/_/  EXTENDED BSD License
//_/_/
//_/_/  Redistribution and use in source and binary forms, with or without
//_/_/  modification, is permitted provided that the following conditions 
//_/_/  are met:
//_/_/
//_/_/  - Redistributions of source code must retain the above copyright 
//_/_/    and collaboration notice, this list of conditions and the following 
//_/_/    disclaimer.
//_/_/
//_/_/  - Redistributions in binary form must reproduce the above copyright 
//_/_/    notice, this list of conditions and the following disclaimer in the 
//_/_/    documentation and/or other materials provided with the distribution.
//_/_/
//_/_/  - Neither the name of its copyright holders nor the names of its 
//_/_/    contributors may be used to endorse or promote products derived from 
//_/_/    this software without specific prior written permission.
//_/_/
//_/_/  - CADIA Clause: The license granted in and to the software under this 
//_/_/    agreement is a limited-use license. The software may not be used in 
//_/_/    furtherance of: (i) intentionally causing bodily injury or severe 
//_/_/    emotional harm to any person; (ii) invading the personal privacy or 
//_/_/    violating the human rights of any person; or (iii) committing or 
//_/_/    preparing for any act of war. 
//_/_/
//_/_/  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS 
//_/_/  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT 
//_/_/  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A 
//_/_/  PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER 
//_/_/  OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, 
//_/_/  EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, 
//_/_/  PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR 
//_/_/  PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF 
//_/_/  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING 
//_/_/  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS 
//_/_/  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
#include "RequestGateway.h"

namespace	cmlabs{

/////////////////////////////////////////////////////////////
// RequestGateway
/////////////////////////////////////////////////////////////

RequestGateway::RequestGateway(uint32 id) {
	cacheFiles = false;
	replyXML = true;
	webServerName = "Server";
	internalAPITitle = "gw/";
	externalAPITitle = "api/";
	sslSupport = false;
	threadID = threadIDClient = 0;
	lastRefID = 0;
	clientSentCount = 0;
	clientReceivedCount = 0;
	execSentCount = 0;
	execReceivedCount = 0;
	shouldContinue = true;
	isRunning = false;
	this->id = id;
	channel = NULL;
	manager = new NetworkManager();
	longReqLimit = 0;
}

RequestGateway::~RequestGateway() {
	stop();

	std::map<uint64, RequestReply*>::iterator i, e;
	mutex.enter(200, __FUNCTION__);
	for (i = requestMap.begin(), e = requestMap.end(); i != e; i++)
		delete(i->second);
	requestMap.clear();
	mutex.leave();

	delete(manager);
}

bool RequestGateway::init(const char* sslCertPath, const char* sslKeyPath) {
	if (sslCertPath && sslKeyPath) {
		if (!manager->setSSLCertificate(sslCertPath, sslKeyPath)) {
			LogPrint(0, LOG_SYSTEM, 0, "SSL Certificate or Key not found in paths: %s and %s!", sslCertPath, sslKeyPath);
			sslSupport = false;
			return false;
		}
		sslSupport = true;
	}

	if (!mutex.enter(200, __FUNCTION__))
		return false;

	if (!ThreadManager::CreateThread(RequestGatewayExecRun, this, threadID, 0)) {
		shouldContinue = false;
		mutex.leave();
		return false;
	}
	if (!ThreadManager::CreateThread(RequestGatewayClientRun, this, threadIDClient, 0)) {
		shouldContinue = false;
		mutex.leave();
		return false;
	}

	mutex.leave();
	return true;
}

bool RequestGateway::setResponseType(const char* type) {
	if (!type || !strlen(type))
		return false;
	replyXML = (strstr(type, "xml") || strstr(type, "XML"));
	return true;
}

bool RequestGateway::setWebServerInfo(const char* name, const char* rootdir, const char* indexfile) {
	if (!name || !rootdir)
		return false;
	webServerName = name;
	this->rootdir = rootdir;
	if (indexfile)
		indexFilename = indexfile;
	return true;
}

bool RequestGateway::setExternalAPILabel(const char* label) {
	if (!label)
		return false;
	externalAPITitle = label;
	return true;
}

bool RequestGateway::setInternalAPILabel(const char* label) {
	if (!label)
		return false;
	internalAPITitle = label;
	return true;
}

bool RequestGateway::setCacheFiles(bool cache) {
	cacheFiles = cache;
	return true;
}

bool RequestGateway::addPort(uint16 port, uint8 encryption, bool enableHTTP, uint32 timeout) {
	if (!mutex.enter(200, __FUNCTION__))
		return false;

	if ((encryption == SSLENC) && !sslSupport) {
		LogPrint(0, LOG_SYSTEM, 0, "SSL encryption not enabled!");
		mutex.leave();
		return false;
	}

	if (enableHTTP) {
		if (!channel) {
			channel = manager->createListener(port, encryption, PROTOCOL_HTTP_SERVER, true, timeout, false, 0, this);
			if (!channel) {
				mutex.leave();
				return false;
			}
		}
		else {
			if (!channel->startListener(0, port, encryption, PROTOCOL_HTTP_SERVER, true, timeout, false)) {
				mutex.leave();
				return false;
			}
		}
	}

	if (!channel) {
		channel = manager->createListener(port, encryption, PROTOCOL_MESSAGE, true, timeout, false, 0, this);
		if (!channel) {
			mutex.leave();
			return false;
		}
	}
	else {
		if (!channel->startListener(0, port, encryption, PROTOCOL_MESSAGE, true, timeout, false)) {
			mutex.leave();
			return false;
		}
	}

	mutex.leave();
	return true;
}


bool RequestGateway::addGateway(uint32 id, std::string addr, uint16 port) {
	return true;
}

bool RequestGateway::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
	if (!mutex.enter(200, __FUNCTION__))
		return false;
	uint64 now = GetTimeNow();
	std::map<uint64,RequestConnection>::iterator i;
	if ( (i = executors.find(conid)) != executors.end()) {
		switch(evt->type) {
			case NETWORKEVENT_DISCONNECT:
				i->second.lastFailTime = now;
				break;
			case NETWORKEVENT_RECONNECT:
				i->second.lastFailTime = 0;
				break;
			default:
				break;
		}
	}
	else if ( (i = clients.find(conid)) != clients.end()) {
		switch(evt->type) {
			case NETWORKEVENT_DISCONNECT:
				i->second.lastFailTime = now;
				break;
			case NETWORKEVENT_RECONNECT:
				i->second.lastFailTime = 0;
				break;
			default:
				break;
		}
	}
	delete(evt);
	mutex.leave();
	return true;
}

bool RequestGateway::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
	if (!msg) return false;
	RequestReply* reply;
	RequestConnection con;
	int64 val1, val2;
	std::map<uint64,RequestConnection>::iterator i;
	uint64 now = GetTimeNow();
	const char* req = msg->getString("URI");
	//LogPrint(0, LOG_SYSTEM, 0, "Gateway received request: '%s'", req);

	if (!mutex.enter(1000)) {
		// ########
		delete(msg);
		return true;
	}

	msg->setRecvTime(now);

	if ( (i=executors.find(conid)) != executors.end()) {
		if (req && (strcmp(req, "ExecutorStatus") == 0) &&
			 msg->getInt("ShortQSize", val1) &&
			 msg->getInt("LongQSize", val2) ) {
			i->second.lastStatusTime = now;
			i->second.shortReqQSize = (uint32) val1;
			i->second.longReqQSize = (uint32) val2;
			delete(msg);
		}
		else {
			execReceivedCount++;
			replyQ.add(msg);
		}
	}
	else if (clients.find(conid) != clients.end()) {
		//LogPrint(0, LOG_SYSTEM, 0, "Gateway received request from client %llu: '%s'", conid, req);
		uint64 origin = channel->getRemoteAddress(conid);
		msg->setString("REMOTE_ADDR", utils::StringFormat("%u.%u.%u.%u:%u", GETIPADDRESSQUADPORT(origin)).c_str());
		reply = new RequestReply();
		reply->origin = conid;
		reply->clientRef = msg->getReference();
		reply->gatewayRef = ++lastRefID;
		msg->setReference(reply->gatewayRef);
		reply->giveRequestMessage(msg);
		reply->setStatus(QUEUED);
		clientReceivedCount++;
		execQ.add(reply);
	}
	else {
		if (req && strcmp(req, "ExecutorConnect") == 0) {
			std::vector<std::string> longRequests = utils::TextListSplit(msg->getString("LongRequests"), ",");
			if (longRequests.size())
				longReqNames = longRequests;
			con.clear();
			con.conID = conid;
			con.lastConTime = now;
			con.location = channel->getRemoteAddress(conid);
			executors[conid] = con;
		}
		else if (req && strcmp(req, "ClientConnect") == 0) {
			con.clear();
			con.conID = conid;
			con.lastConTime = now;
			con.location = channel->getRemoteAddress(conid);
			clients[conid] = con;
		}
		delete(msg);
	}
	mutex.leave();
	return true;
}

bool RequestGateway::receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
//	const char* requestString = req->getRequest();
	const char* requestString = req->getURI();
	if (!requestString || !strlen(requestString))
		return false;

	requestString++; // skip past the first /

	HTTPReply* replyPage = NULL;
	//char* exttype;

	if (!mutex.enter(1000, __FUNCTION__))
		return false;

	//uint64 start = GetTimeNow();
	if (rootdir.length()) {
		//exttype = new char[1024];
		//utils::strcpyavail(exttype, "text/html", 1024, true);
		if ( (!requestString || !strlen(requestString)) && indexFilename.length()) {
			replyPage = new HTTPReply();
			replyPage->createFromFile(GetTimeNow(), "Psyclone", req->ifModifiedSince, true, cacheFiles, (rootdir+indexFilename).c_str());
			//printf("*** Read default file: %.3fms\n", (double)GetTimeAge(start)/1000.0);
			//if (!(html = utils::ReadAFileString(rootdir+indexFilename)).size())
			//	html = utils::StringFormat("Could read index file...");
		}
		else if (strstr(requestString, internalAPITitle.c_str()) == requestString) {
			callInternalAPI(requestString+internalAPITitle.length(), req, channel, conid);
			//printf("*** Internal API: %.3fms\n", (double)GetTimeAge(start)/1000.0);
			delete(req);
			mutex.leave();
			return true;
		}
		else if (strstr(requestString, externalAPITitle.c_str()) == requestString) {
			callExternalAPI(requestString+externalAPITitle.length(), req, channel, conid);
			//printf("*** External API: %.3fms\n", (double)GetTimeAge(start)/1000.0);
			delete(req);
			mutex.leave();
			return true;
		}
		else { // if (strstr(requestString, "html/") == requestString) {
			replyPage = new HTTPReply();
			replyPage->createFromFile(GetTimeNow(), "Psyclone", req->ifModifiedSince, true, cacheFiles, (rootdir+requestString).c_str());
			//printf("*** Read file '%s': %.3fms\n", requestString, (double)GetTimeAge(start)/1000.0);
			//const char* filename = requestString;
			//if ( utils::TextEndsWith(filename, ".html", false) || utils::TextEndsWith(filename, ".txt", false) ) {
			//	if (!(html = utils::ReadAFileString(rootdir+filename)).size())
			//		html = utils::StringFormat("Could read file '%s'...", filename);
			//}
			//else {
			//	if ( !(data = utils::ReadAFile((rootdir+filename).c_str(), datasize, true)))
			//		html = utils::StringFormat("Could read file '%s'...", filename);
			//	else {
			//		if (!html::GetMimeType(exttype, filename, 1024))
			//			html = utils::StringFormat("Unsupported mime type for file '%s'...", filename);
			//	}
			//}
		}
		//if (!replyPage) {
		//	if (data) {
		//		replyPage = new HTTPReply();
		//		replyPage->createPage(HTTP_OK, GetTimeNow(), "Psyclone", 0, true, cacheFiles, exttype, data, datasize);
		//	}
		//	else if (html.length()) {
		//		replyPage = new HTTPReply();
		//		replyPage->createPage(HTTP_OK, GetTimeNow(), "Psyclone", 0, true, cacheFiles, exttype, html.c_str());
		//	}
		//}
		if (replyPage) {
			mutex.leave();
			channel->sendHTTPReply(replyPage, conid);
			delete(replyPage);
			//delete [] data;
			//delete [] exttype;
			delete(req);
			return true;
		}
		else {
			//delete [] data;
			//delete [] exttype;
		}
	}
	mutex.leave();
	delete(req);
	//printf("*** No reply: %.3fms\n", (double)GetTimeAge(start)/1000.0);
	return false;

}

bool RequestGateway::callExternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
	RequestConnection con;
	std::map<uint64,RequestConnection>::iterator i;
	RequestReply* reply;
	uint64 now = GetTimeNow();
	DataMessage* msg;
	uint64 ref = utils::Ascii2Uint64(req->getParameter("Reference"));

	if ( (i=webClients.find(conid)) == webClients.end()) {
		con.clear();
		con.conID = conid;
		con.lastConTime = now;
		webClients[conid] = con;
	}

	// convert to DataMessage
	msg = req->convertToMessage();
//	msg->setString("REQUEST", apiName);
	msg->setString("URI", apiName);
	msg->setString("REMOTE_ADDR", utils::StringFormat("%u.%u.%u.%u:%u", GETIPADDRESSQUADPORT(req->source)).c_str());
	reply = new RequestReply();
	reply->origin = conid;
	reply->gatewayRef = ++lastRefID;
	if (ref)
		reply->clientRef = ref;
	else
		reply->clientRef = reply->gatewayRef;
	//msg->setReference(reply->clientRef);
	msg->setReference(reply->gatewayRef);
	reply->giveRequestMessage(msg);
	reply->setStatus(QUEUED);
	execQ.add(reply);

	return true;
}

bool RequestGateway::callInternalAPI(const char* apiName, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
	HTTPReply* reply = new HTTPReply();
	std::string xml, subXML, json, execJSON, clientJSON, webClientJSON;
	std::map<uint64,RequestConnection>::iterator i,e;
	uint32 clientCount = 0, execCount = 0, webClientCount = 0;
	uint64 now = GetTimeNow();
	uint32 ms1 = 1000, ms2 = 10000, ms3 = 60000;
	double lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3;
	double svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3;

	if (stricmp(apiName, "status") == 0) {
		for (i=executors.begin(), e=executors.end(); i!=e; i++) {
			if (!i->second.lastFailTime) {
				if (replyXML) {
					subXML += utils::StringFormat("<executor id=\"%llu\" location=\"%u.%u.%u.%u:%u\" shortQueue=\"%u\" longQueue=\"%u\" count=\"%u\" uptime=\"%s\" />\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.shortReqQSize,
						i->second.longReqQSize,
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
						);
				}
				else {
					if (execJSON.length())
						execJSON += ",";
					execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortQueue\":%u, \"longQueue\":%u, \"count\":%u, \"uptime\":\"%s\" }\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.shortReqQSize,
						i->second.longReqQSize,
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
						);
				}
				execCount++;
			}
		}
		for (i=clients.begin(), e=clients.end(); i!=e; i++) {
			if (!i->second.lastFailTime) {
				//shortAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
				//longAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
				if (replyXML) {
					subXML += utils::StringFormat("<client id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" />\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
						);
				}
				else {
					if (clientJSON.length())
						clientJSON += ",";
					clientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\" }\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
						);
				}
				clientCount++;
			}
		}
		for (i = webClients.begin(), e = webClients.end(); i != e; i++) {
			if (!i->second.lastFailTime) {
				//shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
				//longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
				if (replyXML) {
					subXML += utils::StringFormat("<webclient id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" />\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
						);
				}
				else {
					if (webClientJSON.length())
						webClientJSON += ",";
					webClientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\" }\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
						);
				}
				webClientCount++;
			}
		}
		shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
		longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
		if (replyXML)
			xml = utils::StringFormat("<status executors=\"%u\" clients=\"%u\" webclients=\"%u\" "
				"shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
				"longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
				">\n%s</status>\n", execCount, clientCount, webClientCount,
				svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
				lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
				subXML.c_str());
		else
			json = utils::StringFormat("{\"executorcount\":%u, \"clientcount\":%u, \"webclientcount\":%u, "
				"\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
				"\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f, "
				"\"executors\":[%s], \"clients\":[%s], \"webclients\":[%s]}\n",
				execCount, clientCount, webClientCount,
				svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
				lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
				execJSON.c_str(), clientJSON.c_str(), webClientJSON.c_str());
	}
	else if (stricmp(apiName, "stats") == 0) {
		svalPerSec1 = scountPerSec1 = svalPerSec2 = scountPerSec2 = svalPerSec3 = scountPerSec3 =
			lvalPerSec1 = lcountPerSec1 = lvalPerSec2 = lcountPerSec2 = lvalPerSec3 = lcountPerSec3 = 0;
		for (i = executors.begin(), e = executors.end(); i != e; i++) {
			if (!i->second.lastFailTime) {
				//shortAvgStatsExec[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
				//longAvgStatsExec[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
				if (replyXML) {
					subXML += utils::StringFormat("<executor id=\"%llu\" location=\"%u.%u.%u.%u:%u\" shortQueue=\"%u\" longQueue=\"%u\" count=\"%u\" uptime=\"%s\" "
						"shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
						"longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.shortReqQSize,
						i->second.longReqQSize,
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
						svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
						lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
						);
				}
				else {
					if (execJSON.length())
						execJSON += ",";
					//execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortQueue\":%u, \"longQueue\":%u, \"count\":%u, \"uptime\":\"%s\" }\n",
					//	i->second.conID,
					//	GETIPADDRESSQUADPORT(i->second.location),
					//	i->second.shortReqQSize,
					//	i->second.longReqQSize,
					//	i->second.count,
					//	PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str()
					//	);
					execJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"shortQueue\":%u, \"longQueue\":%u, \"count\":%u, \"uptime\":\"%s\", "
						"\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
						"\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.shortReqQSize,
						i->second.longReqQSize,
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
						svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
						lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
						);
				}
				execCount++;
			}
		}
		for (i = clients.begin(), e = clients.end(); i != e; i++) {
			if (!i->second.lastFailTime) {
				//shortAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
				//longAvgStatsClients[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
				if (replyXML) {
					subXML += utils::StringFormat("<client id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
						"shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
						"longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
						svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
						lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
						);
				}
				else {
					if (clientJSON.length())
						clientJSON += ",";
					clientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
						"\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
						"\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
						svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
						lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
						);
				}
				clientCount++;
			}
		}
		for (i = webClients.begin(), e = webClients.end(); i != e; i++) {
			if (!i->second.lastFailTime) {
				//shortAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
				//longAvgStatsWeb[i->second.conID].getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
				if (replyXML) {
					subXML += utils::StringFormat("<webclient id=\"%llu\" location=\"%u.%u.%u.%u:%u\" count=\"%u\" uptime=\"%s\" "
						"shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
						"longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" />\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
						svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
						lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
						);
				}
				else {
					if (webClientJSON.length())
						webClientJSON += ",";
					webClientJSON += utils::StringFormat("{\"id\":%llu, \"location\":\"%u.%u.%u.%u:%u\", \"count\":%u, \"uptime\":\"%s\", "
						"\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
						"\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f }\n",
						i->second.conID,
						GETIPADDRESSQUADPORT(i->second.location),
						i->second.count,
						PrintTimeDifString(GetTimeAge(i->second.lastConTime)).c_str(),
						svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
						lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3
						);
				}
				webClientCount++;
			}
		}
		shortAvgStats.getThroughputMulti(ms1, ms2, ms3, svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3, now);
		longAvgStats.getThroughputMulti(ms1, ms2, ms3, lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3, now);
		if (replyXML)
			xml = utils::StringFormat("<status executors=\"%u\" clients=\"%u\" webclients=\"%u\" "
			"shortBytesPerSec1=\"%.3f\" shortCountPerSec1=\"%.3f\" shortBytesPerSec10=\"%.3f\" shortCountPerSec10=\"%.3f\" shortBytesPerSec60=\"%.3f\" shortCountPerSec60=\"%.3f\" "
			"longBytesPer1Sec=\"%.3f\" longCountPer1Sec=\"%.3f\" longBytesPer10Sec=\"%.3f\" longCountPer10Sec=\"%.3f\" longBytesPer60Sec=\"%.3f\" longCountPer60Sec=\"%.3f\" "
			">\n%s</status>\n", execCount, clientCount, webClientCount,
			svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
			lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
			subXML.c_str());
		else
			json = utils::StringFormat("{\"executorcount\":%u, \"clientcount\":%u, \"webclientcount\":%u, "
			"\"shortBytesPerSec1\": %.3f, \"shortCountPerSec1\": %.3f, \"shortBytesPerSec10\": %.3f, \"shortCountPerSec10\": %.3f, \"shortBytesPerSec60\": %.3f, \"shortCountPerSec60\": %.3f, "
			"\"longBytesPer1Sec\": %.3f, \"longCountPer1Sec\": %.3f, \"longBytesPer10Sec\": %.3f, \"longCountPer10Sec\": %.3f, \"longBytesPer60Sec\": %.3f, \"longCountPer60Sec\": %.3f, "
			"\"executors\":[%s], \"clients\":[%s], \"webclients\":[%s]}\n",
			execCount, clientCount, webClientCount,
			svalPerSec1, scountPerSec1, svalPerSec2, scountPerSec2, svalPerSec3, scountPerSec3,
			lvalPerSec1, lcountPerSec1, lvalPerSec2, lcountPerSec2, lvalPerSec3, lcountPerSec3,
			execJSON.c_str(), clientJSON.c_str(), webClientJSON.c_str());
	}
	else if (stricmp(apiName, "sizeechotest") == 0) {
		// size = contentlength - 40 (as we add about 40 chars of XML below)
		uint32 size = (req->contentLength < 50) ? 10 : req->contentLength - 40;
		uint32 n;
		char* echocontent = new char[size+1];
		for (n=0; n<size; n++)
			echocontent[n] = 48 + (n%10);
		echocontent[n] = 0;
		if (replyXML)
			xml = utils::StringFormat("<echosizetest size=\"%u\">%s</echosizetest>", size, echocontent);
		else
			json = utils::StringFormat("{\"size\":%u, \"content\":\"%s\"}\n", size, echocontent);
		delete [] echocontent;
	}
	else {
		if (replyXML)
			xml = utils::StringFormat("<error code=\"-1\" text=\"API does not exist: %s\" />", apiName);
		else
			json = utils::StringFormat("{\"errorcode\":-1, \"text\":\"API does not exist: %s\"}\n", apiName);
	}

	if (replyXML)
		reply->createPage(HTTP_OK, GetTimeNow(), webServerName.c_str(), 0, true, false, "text/xml", xml.c_str());
	else
		reply->createPage(HTTP_OK, GetTimeNow(), webServerName.c_str(), 0, true, false, "application/json", json.c_str());

	channel->sendHTTPReply(reply, conid);
	delete(reply);
	return true;
}

bool RequestGateway::setLongRequestLimit(uint32 limit) {
	longReqLimit = limit;
	return true;
}

bool shortQIsSmaller(const RequestConnection& r1, const RequestConnection& r2) {
   return (r1.shortReqQSize < r2.shortReqQSize);
}

bool longQIsSmaller(const RequestConnection& r1, const RequestConnection& r2) {
   return (r1.longReqQSize < r2.longReqQSize);
}

std::list<RequestConnection> RequestGateway::getTopExecutors(std::string requestString, uint32 reqSize, bool& isLongReq) {
	isLongReq = false;
	if (longReqLimit)
		isLongReq = (reqSize >= longReqLimit);
	if (!isLongReq) {
		std::vector<std::string>::iterator il = longReqNames.begin(), el = longReqNames.end();
		while (il != el) {
			if (utils::stristr(requestString.c_str(), il->c_str())) {
				isLongReq = true;
				break;
			}
			il++;
		}
	}
	std::list<RequestConnection> executorsList;
	std::map<uint64,RequestConnection>::iterator i,e;
	for (i=executors.begin(), e=executors.end(); i!=e; i++)
		executorsList.push_back(i->second);
	if (isLongReq) {
		executorsList.sort(longQIsSmaller);
		//printf("--- Long EXELIST [%u] ---\n", executorsList.size());
	}
	else {
		executorsList.sort(shortQIsSmaller);
		//std::sort(executorsList.begin(), executorsList.end(), shortQIsSmaller);
		//printf("--- Short EXELIST [%u] ---\n", executorsList.size());
	}

	//uint32 n = 0;
	//std::list<RequestConnection>::iterator ii,ee;
	//for (ii=executorsList.begin(), ee=executorsList.end(); ii!=ee; ii++) {
	//	printf("    [%u] [%llu]  SQ: %u  LQ: %u\n", n, (*ii).conID, (*ii).shortReqQSize, (*ii).longReqQSize);
	//	fflush(stdout);
	//}

	return executorsList;
}

bool RequestGateway::sendToExecutor(RequestReply* reply) {
	std::list<RequestConnection> topExecutors;
	std::list<RequestConnection>::iterator i, e;
	requestMap[reply->gatewayRef] = reply;
	DataMessage *replyMsg;

	DataMessage* reqMsg = reply->peekRequestMessage();
	if (!reqMsg) {
		replyMsg = new DataMessage();
		replyMsg->setReference(reply->clientRef);
		if (replyXML)
			replyMsg->setString("XML", "<error>No input data available</error>");
		else
			replyMsg->setString("JSON", "{\"error\": \"No input data available\"}\n");
		replyQ.add(replyMsg);
		return false;
	}
	const char* reqStr = reqMsg->getString("URI");
	if (!reqStr)
		reqStr = reqMsg->getString("REQUEST");
	if (!reqStr) {
		replyMsg = new DataMessage();
		replyMsg->setReference(reply->clientRef);
		if (replyXML)
			replyMsg->setString("XML", "<error>No request data available</error>");
		else
			replyMsg->setString("JSON", "{\"error\": \"No request data available\"}\n");
		replyQ.add(replyMsg);
		return false;
	}

	topExecutors = getTopExecutors(reqStr, reqMsg->getSize(), reply->isLongReq);

	if (!topExecutors.size()) {
		LogPrint(0, LOG_SYSTEM, 0, "No executors found!");
		reply->setStatus(FAILED);
		replyMsg = new DataMessage();
		replyMsg->setReference(reply->clientRef);
		if (replyXML)
			replyMsg->setString("XML", "<error>No processors available</error>");
		else
			replyMsg->setString("JSON", "{\"error\": \"No processors available\"}\n");
		replyQ.add(replyMsg);
		return false;
	}

	bool wasSent = false;
	for (i=topExecutors.begin(), e=topExecutors.end(); i!=e; i++) {
		if (channel->sendMessage(reply->peekRequestMessage(), (*i).conID)) {
			executors[(*i).conID].count++;
			reply->processor = (*i).conID;
			//printf("--- GW -> EX[%llu] ---\n", (*i).conID);
			//fflush(stdout);
			wasSent = true;
			break;
		}
	}
	if (wasSent) {
		reply->setStatus(SENT);
		execSentCount++;
	}
	else {
		reply->setStatus(SERVERERROR);
		// #########
		// reply to client: server error
		return false;
	}
	return true;
}

bool RequestGateway::replyToClient(DataMessage* msg) {
	uint64 gwRef = msg->getReference();
	std::map<uint64,RequestReply*>::iterator i = requestMap.find(gwRef);
	std::map<uint64,RequestConnection>::iterator con;

	if (i == requestMap.end()) {
		// #############
		delete(msg);
		return false;
	}
	RequestReply* reply = i->second;

	// Add Gateway stats
	//MovingAverage* avg;
	uint32 reqSize = reply->peekRequestMessage()->getSize();
	if (reply->isLongReq) {
		longAvgStats.add(reqSize);
		//avg = &(longAvgStatsExec[reply->processor]);
		//avg->add(reqSize);
	}
	else {
		shortAvgStats.add(reqSize);
		//avg = &(shortAvgStatsExec[reply->processor]);
		//avg->add(reqSize);
	}

	if ( (con=webClients.find(reply->origin)) != webClients.end()) {
		//if (reply->isLongReq)
		//	avg = &(longAvgStatsWeb[reply->origin]);
		//else
		//	avg = &(shortAvgStatsWeb[reply->origin]);
		//avg->add(reqSize);
		HTTPReply* httpReply = new HTTPReply();
		const char* xml, *html, *text, *data, *mime, *json, *cache, *options, *origin;
		int64 resultCode = 0;
		msg->getInt("ResultCode", resultCode);
		if (!resultCode) resultCode = HTTP_OK;
		cache = msg->getString("CACHE");
		bool cacheable = (cache && (utils::stristr(cache, "yes") == cache));
		uint32 size;
		// ########## Add reference and reply to HTTP reply #######
		if (xml = msg->getString("XML"))
			httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "text/xml", xml);
		else if (json = msg->getString("JSON"))
			httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "application/json", json);
		else if (html = msg->getString("HTML"))
			httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "text/html", html);
		else if (text = msg->getString("TEXT"))
			httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, "text/text", html);
		else if ( (data = msg->getData("BINARY", size)) && size && (mime = msg->getString("MIMETYPE")) )
			httpReply->createPage((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, true, cacheable, mime, data, size);
		else if ( (options = msg->getString("OPTIONS")) && (origin = msg->getString("ORIGIN")) )
			httpReply->createOptionsResponse((uint8)resultCode, GetTimeNow(), webServerName.c_str(), 0, origin, options);
		else
			httpReply->createErrorPage(HTTP_SERVER_NOREPLY, GetTimeNow(), webServerName.c_str(), true);

		if (channel->sendHTTPReply(httpReply, reply->origin)) {
			con->second.count++;
			delete(httpReply);
			delete(msg);
			requestMap.erase(i);
			delete(reply);
			return true;
		}
		else {
			delete(httpReply);
			delete(msg);
			requestMap.erase(i);
			delete(reply);
			return false;
		}
	}
	else if ( (con=clients.find(reply->origin)) != clients.end()) {
		//if (reply->isLongReq)
		//	avg = &(longAvgStatsClients[reply->origin]);
		//else
		//	avg = &(shortAvgStatsClients[reply->origin]);
		//avg->add(reqSize);
		msg->setReference(reply->clientRef);
		msg->setFrom((uint32)reply->processor);
		if (channel->sendMessage(msg, reply->origin)) {
			con->second.count++;
			clientSentCount++;
			// success
			delete(msg);
			requestMap.erase(i);
			delete(reply);
			return true;
		}
		else {
			delete(msg);
			requestMap.erase(i);
			delete(reply);
			return false;
		}
	}

	// else it failed
	// ########
	delete(msg);
	requestMap.erase(i);
	delete(reply);
	return false;
}


bool RequestGateway::runExec() {
	RequestReply* reply;
	isRunning = true;

	while (shouldContinue) {
		while (execQ.waitForAndTakeFirst(reply,50))
			sendToExecutor(reply);
	}
	isRunning = false;
	return true;
}

bool RequestGateway::runClient() {
	DataMessage* msg;
	isRunning = true;

	while (shouldContinue) {
		while (msg = replyQ.waitForAndTakeFirst(50))
			replyToClient(msg);
	}
	isRunning = false;
	return true;
}

THREAD_RET THREAD_FUNCTION_CALL RequestGatewayExecRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((RequestGateway*)arg)->runExec() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL RequestGatewayClientRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((RequestGateway*)arg)->runClient() ? 0 : 1));
}



















bool RequestGateway::UnitTest() {
	uint16 gatewayCount = 1;
	uint16 executorCount = 3;
	uint16 clientCount = 3;
	uint16 requestCount = 30;
	uint16 n, m;

	//uint8 encryption = SSLENC;
	uint8 encryption = NOENC;

	DataMessage *msg;
	RequestReply* reply;
	std::list<RequestReply*> replyList;
	RequestStatus status;

	RequestGateway* gateway;
	TestRequestExecutor* exec;
	RequestClient* client;

	std::vector<RequestGateway*> gateways;
	std::vector<TestRequestExecutor*> executors;
	std::vector<RequestClient*> clients;

	std::string sslCertPath = "../cert";
	std::string sslKeyPath = "../pkey";

	// Start N gateways
	for (n=0; n<gatewayCount; n++) {
		gateway = new RequestGateway(n);
		if (encryption == SSLENC) {
			if (gateway->init(sslCertPath.c_str(), sslKeyPath.c_str()))
				gateways.push_back(gateway);
			else {
				LogPrint(0, LOG_SYSTEM, 0, "Couldn't initialise gateway %u...", n);
				return false;
			}
		}
		else {
			if (gateway->init())
				gateways.push_back(gateway);
			else {
				LogPrint(0, LOG_SYSTEM, 0, "Couldn't initialise gateway %u...", n);
				return false;
			}
		}
		if (!gateway->addPort(2000+n, encryption)) {
			LogPrint(0, LOG_SYSTEM, 0, "Couldn't initialise gateway %u port %u...", n, 2000+n);
			return false;
		}
	}

	// Start N executors
	for (n=0; n<executorCount; n++) {
		exec = new TestRequestExecutor();
		exec->addLongRequestName("longtest");
		for (m=0; m<gatewayCount; m++) {
			if (!exec->addGateway(m, "localhost", 2000+m, encryption)) {
				LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway %u to executor %u...", m, n);
				return false;
			}
		}
		executors.push_back(exec);
	}

	LogPrint(0, LOG_SYSTEM, 0, "Gateway(s) and Executor(s) set up and started");
	utils::Sleep(1000);

	// Start N clients
	for (n=0; n<clientCount; n++) {
		client = new RequestClient();
		for (m=0; m<gatewayCount; m++) {
			if (!client->addGateway(m, "localhost", 2000+m, encryption)) {
				LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway %u to client %u...\n", m, n);
				return false;
			}
		}
		clients.push_back(client);
	}

	// Send asynchronous binary requests
	for (n=0; n<clientCount; n++) {
		client = clients[n];
		for (m=0; m<requestCount/2; m++) {
			msg = new DataMessage();
			msg->setString("URI", "shorttest");
			if (reply = client->postRequest(msg)) {
				reply->customRef = 1;
				replyList.push_back(reply);
			}
			else {
				LogPrint(0, LOG_SYSTEM, 0, "Couldn't post short query %u via client %u...\n", m, n);
				delete(msg);
				return false;
			}
			//printf("Posted short query %u via client %u...\n", m, n);
			msg = new DataMessage();
			msg->setString("URI", "longtest");
			if (reply = client->postRequest(msg)) {
				reply->customRef = 2;
				replyList.push_back(reply);
			}
			else {
				LogPrint(0, LOG_SYSTEM, 0, "Couldn't post long query %u via client %u...\n", m, n);
				delete(msg);
				utils::Sleep(50000);
				return false;
			}
			//printf("Posted long query %u via client %u...\n", m, n);
		}
		utils::Sleep(200);
	}
	//printf("Finished posting, now waiting...\n");

	// Wait for results of asynchronous requests

	std::list<RequestReply*>::iterator i,e;

	uint64 startTime = GetTimeNow();
	uint32 stillWaiting, failed, timeout, errors, success, total;
	stillWaiting = failed = timeout = errors = success = total = 0;
	while (true) {
		stillWaiting = failed = timeout = errors = success = total = 0;
		for (i=replyList.begin(),e=replyList.end(); i!=e; i++) {
			reply = *i;
			status = reply->getStatus();
			total++;
			switch(status) {
				case SUCCESS:
					//if (reply->customRef == 2) {
					//	longStats.add((double)reply->getRequestDuration());
					//	//printf("--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
					//}
					//else if (reply->customRef == 1) {
					//	shortStats.add((double)reply->getRequestDuration());
					//	//printf("--- [%u] --- Short request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
					//}
					success++;
					break;
				case FAILED:
					failed++;
					break;
				case LOCALERROR:
				case NETWORKERROR:
				case SERVERERROR:
					errors++;
					break;
				case TIMEOUT:
					timeout++;
					break;
				default:
					stillWaiting++;
					break;
			}
		}
		if (!stillWaiting || GetTimeAgeMS(startTime) > 60000)
			break;
		//if (stillWaiting != total) {
			LogPrint(0, LOG_SYSTEM, 0, "----- [%u/%u] success: %u  failed: %u  timeout: %u  errors: %u\n",
				stillWaiting, total, success, failed, timeout, errors);
			//printf("--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
		//}
		utils::Sleep(200);
	}
	if (total == success) {
		Stats shortStats, longStats;
		for (i=replyList.begin(),e=replyList.end(); i!=e; i++) {
			reply = *i;
			if (reply->customRef == 2) {
				longStats.add((double)reply->getRequestDuration());
				LogPrint(0, LOG_SYSTEM, 0, "--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
			}
			else if (reply->customRef == 1) {
				shortStats.add((double)reply->getRequestDuration());
				LogPrint(0, LOG_SYSTEM, 0, "--- [%u] --- Short request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
			}
		}
		LogPrint(0, LOG_SYSTEM, 0, "[%u] success - short stats: %.3f (%.3f)  long stats: %.3f (%.3f)\n",
			success, shortStats.getAverage()/1000.0, shortStats.getStdDev()/1000.0, longStats.getAverage()/1000.0, longStats.getStdDev()/1000.0);
	}
	else {
		LogPrint(0, LOG_SYSTEM, 0, "[%u/%u] success: %u  failed: %u  timeout: %u  errors: %u\n",
			stillWaiting, total, success, failed, timeout, errors);
	}

	utils::Sleep(500000);
	// Send synchronous HTTP requests
	// #########################

	std::vector<RequestGateway*>::iterator gi, ge;
	for (gi = gateways.begin(), ge = gateways.end(); gi != ge; gi++)
		delete(*gi);
	std::vector<TestRequestExecutor*>::iterator ei, ee;
	for (ei = executors.begin(), ee = executors.end(); ei != ee; ei++)
		delete(*ei);
	std::vector<RequestClient*>::iterator ci, ce;
	for (ci = clients.begin(), ce = clients.end(); ci != ce; ci++)
		delete(*ci);

	return true;
}



}
