

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 "NetworkManager.h"

namespace	cmlabs{



/////////////////////////////////////////////////////////////
// Network Manager
/////////////////////////////////////////////////////////////

NetworkThread::NetworkThread(NetworkChannel* parent, uint64 id) {
	this->parent = parent;
	this->id = id;

	shouldContinue = true;
	port = 0;
	isRunning = false;
	autoreconnect = false;
	isAsync = true;
	autoProtocols = 0;
	autoProtocolTimeout = 0;
	defaultProtocol = 0;

	con = NULL;
	listener = NULL;
	lastRequest = NULL;
}

NetworkThread::~NetworkThread() {
	autoreconnect = false;
	shouldContinue = false;
	isRunning = false;

	if (listener != NULL)
		delete(listener);
	listener = NULL;
	if (con != NULL)
		delete(con);
	con = NULL;
	if (lastRequest != NULL)
		delete(lastRequest);
	lastRequest = NULL;

}


THREAD_RET THREAD_FUNCTION_CALL NetworkManager::NetworkManagerRun(THREAD_ARG arg) {
	thread_ret_val(0);
}

NetworkManager::NetworkManager() {
	lastChannelID = 0;
	lastConnectionID = 0;
	udpOutputCon = NULL;
}

NetworkManager::~NetworkManager() {

	// Simply delete listeners map as they are stored in channels anyway
	listeners.clear();
	// Simply delete channelsByConnection map as they are stored in channels anyway
	channelsByConnection.clear();
	// Shutdown all channels
	std::map<uint32, NetworkChannel*>::iterator it, itEnd;
	for (it = channels.begin(), itEnd = channels.end(); it != itEnd; ++it) {
		if (it->second != NULL) {
			it->second->shutdown();
			delete(it->second);
		}
	}
	channels.clear();
	if (udpOutputCon)
		delete(udpOutputCon);
	udpOutputCon = NULL;
}

bool NetworkManager::setSSLCertificate(const char* sslCertPath, const char* sslKeyPath) {
	this->sslCertPath = sslCertPath;
	this->sslKeyPath = sslKeyPath;
	return true;
}


NetworkChannel* NetworkManager::createListener(uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint32 protocolTimeout, bool isDefaultProtocol, uint32 channelID, NetworkReceiver* recv) {
	bool createdConnection = false;
	// First check if the port is already in use
	NetworkChannel* channel = listeners[port];
	if (channel) {
		channelID = channel->cid;
	}
	else {
		if (channelID > 0)
			channel = channels[channelID];
		if (channel == NULL) {
			channelID = ++lastChannelID;
			while (getConnection(channelID))
				channelID = ++lastChannelID;
			channel = new NetworkChannel(this);
			channel->cid = channelID;
			channels[channelID] = channel;
			listeners[port] = channel;
			createdConnection = true;
		}
	}
	// We have got a connection, check that we can bind
	if (!channel->startListener(channelID, port, encryption, protocol, isAsync, protocolTimeout, isDefaultProtocol)) {
		if (createdConnection) {
			listeners.erase(port);
			channels.erase(channelID);
			delete(channel);
		}
		return NULL;
	}

	if (recv)
		channel->setNewReceiver(recv);
	return channel;
}

bool NetworkManager::stopListener(uint16 port, uint8 protocol) {
	NetworkChannel* channel = listeners[port];
	if (!channel)
		return false;
	return channel->stopListener(port, protocol);
}

NetworkChannel* NetworkManager::createTCPConnection(const char* addr, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint64& location) {
	bool createdConnection = false;
	NetworkChannel* channel = NULL;
	if (channelID > 0)
		channel = channels[channelID];
	if (channel == NULL) {
		channelID = ++lastChannelID;
		while (getConnection(channelID))
			channelID = ++lastChannelID;
		channel = new NetworkChannel(this);
		channel->cid = channelID;
		channels[channelID] = channel;
		createdConnection = true;
	}

	// We have got a connection, check that we can bind
	if ( (conid = channel->createTCPConnection(addr, port, encryption, protocol, isAsync, autoreconnect, location)) == 0) {
		if (createdConnection) {
			channels.erase(channelID);
			delete(channel);
		}
		return NULL;
	}
	channelsByConnection[conid] = channel;

	if (recv)
		channel->setNewReceiver(recv);

	LogPrint(0,LOG_NETWORK,2,"New connection %llu to %s:%u created...",
		conid, addr, port);

	return channel;
}

NetworkChannel* NetworkManager::createTCPConnection(uint64 location, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid) {
	bool createdConnection = false;
	NetworkChannel* channel = NULL;
	if (channelID > 0)
		channel = channels[channelID];
	if (channel == NULL) {
		// Try creating the connection first
		TCPConnection* con = new TCPConnection();
		if (!con->connect(location)) {
			delete(con);
			return NULL;
		}
		channelID = ++lastChannelID;
		while (getConnection(channelID))
			channelID = ++lastChannelID;
		channel = new NetworkChannel(this);
		channel->cid = channelID;
		if ( (conid = channel->startConnection(con, protocol, isAsync, autoreconnect)) == 0) {
			delete(channel);
			return NULL;
		}
		channels[channelID] = channel;
		createdConnection = true;
	}
	else {
		// We have got a channel, check that we can connect
		if ( (conid = channel->createTCPConnection(location, encryption, protocol, isAsync, autoreconnect)) == 0) {
			if (createdConnection) {
				channels.erase(channelID);
				delete(channel);
			}
			return NULL;
		}
	}
	channelsByConnection[conid] = channel;

	if (recv)
		channel->setNewReceiver(recv);

	LogPrint(0,LOG_NETWORK,2,"New connection %llu to %u.%u.%u.%u:%u created...",
		conid, GETIPADDRESSQUADPORT(location));

	return channel;
}

NetworkChannel* NetworkManager::createTCPConnection(const uint32* addresses, uint16 addressCount, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid, uint64& location) {
	bool createdConnection = false;
	NetworkChannel* channel = NULL;
	if (channelID > 0)
		channel = channels[channelID];
	if (channel == NULL) {
		channelID = ++lastChannelID;
		while (getConnection(channelID))
			channelID = ++lastChannelID;
		channel = new NetworkChannel(this);
		channel->cid = channelID;
		channels[channelID] = channel;
		createdConnection = true;
	}

	// We have got a connection, check that we can bind
	if ( (conid = channel->createTCPConnection(addresses, addressCount, port, encryption, protocol, isAsync, autoreconnect, location)) == 0) {
		if (createdConnection) {
			channels.erase(channelID);
			delete(channel);
		}
		return NULL;
	}
	channelsByConnection[conid] = channel;

	if (recv)
		channel->setNewReceiver(recv);

	LogPrint(0,LOG_NETWORK,2,"New connection %llu to %u.%u.%u.%u:%u created...",
		conid, GETIPADDRESSQUADPORT(location));

	return channel;
}

NetworkChannel* NetworkManager::createUDPConnection(uint16 port, uint8 protocol, bool isAsync, bool autoreconnect, uint32 channelID, NetworkReceiver* recv, uint64& conid) {
	bool createdConnection = false;
	NetworkChannel* channel = NULL;
	if (channelID > 0)
		channel = getConnection(channelID);
	if (channel == NULL) {
		channelID = ++lastChannelID;
		while (getConnection(channelID))
			channelID = ++lastChannelID;
		channel = new NetworkChannel(this);
		channel->cid = channelID;
		channels[channelID] = channel;
		createdConnection = true;
	}

	// We have got a connection, check that we can bind
	if ( (conid = channel->createUDPConnection(port, protocol, isAsync, autoreconnect)) == 0) {
		if (createdConnection) {
			channels.erase(channelID);
			delete(channel);
		}
		return NULL;
	}
	udpListeners[port] = channel;
	channelsByConnection[conid] = channel;

	if (recv)
		channel->setNewReceiver(recv);
	return channel;
}

bool NetworkManager::endConnection(uint64 conid) {
	NetworkChannel* channel = channelsByConnection[conid];
	if (!channel)
		return false;
	return channel->endConnection(conid);
}

bool NetworkManager::endUDPConnection(uint16 port) {
	NetworkChannel* channel = udpListeners[port];
	if (!channel)
		return false;
	return channel->endUDPConnection(port);
}

uint64 NetworkManager::addConnection(NetworkChannel* channel) {
	uint64 conid = ++lastConnectionID;
	channelsByConnection[conid] = channel;
	return conid;
}

uint8 NetworkManager::getConnectionType(uint64 conid) {
	NetworkChannel* channel = channelsByConnection[conid];
	if (!channel)
		return 0;
	return channel->getConnectionType(conid);
}

uint64 NetworkManager::getRemoteAddress(uint64 conid) {
	NetworkChannel* channel = channelsByConnection[conid];
	if (!channel)
		return 0;
	return channel->getRemoteAddress(conid);
}

NetworkChannel* NetworkManager::getConnection(uint64 conID) {
	return channelsByConnection[conID];
}

NetworkChannel* NetworkManager::getTCPConnectionByPort(uint16 port) {
	return listeners[port];
}

NetworkChannel* NetworkManager::getUDPConnectionByPort(uint16 port) {
	return udpListeners[port];
}

bool NetworkManager::sendUDPMessage(DataMessage* msg, uint64 destination) {
	if (!udpOutputConMutex.enter(3000, __FUNCTION__))
		return false;

	if (!udpOutputCon) {
		udpOutputCon = new UDPConnection();
		if (!udpOutputCon->initForOutputOnly()) {
			delete(udpOutputCon);
			udpOutputCon = NULL;
			udpOutputConMutex.leave();
			return false;
		}
	}
	
	bool res = MessageProtocol::SendMessage(udpOutputCon, msg, destination);
	udpOutputConMutex.leave();
	return res;
}


HTTPReply* NetworkManager::makeHTTPRequest(const char* url, uint32 timeout, const char* content, uint32 contentSize) {
	// http://localhost:8000/getstatus.php?count=10

	std::string protocolString = html::GetProtocolFromURL(url);
	int8 encryption = NOENC;
	if (protocolString == "http") {}
	else if (protocolString == "https")
		encryption = SSLENC;
	else
		return HTTPReply::CreateErrorReply(HTTP_NOT_IMPLEMENTED);

	std::string hostString = html::GetHostFromURL(url);
	if (!hostString.size())
		return HTTPReply::CreateErrorReply(HTTP_MALFORMED_URL);

	uint16 port = html::GetPortFromURL(url);
	if (!port) {
		if (encryption == SSLENC)
			port = 443;
		else
			port = 80;
	}

	std::string uriString = html::GetURIFromURL(url);
	if (!uriString.size())
		uriString = "/";

	HTTPRequest* req = new HTTPRequest();
	if (content && contentSize)
		req->createRequest(HTTP_POST, "", uriString.c_str(), content, contentSize, false, 0);
	else
		req->createRequest(HTTP_GET, "", uriString.c_str(), NULL, 0, false, 0);
	HTTPReply* reply = makeHTTPRequest(req, hostString.c_str(), port, encryption, timeout);
	delete(req);
	return reply;
}

HTTPReply* NetworkManager::makeHTTPRequest(HTTPRequest* req, const char* addr, uint16 port, uint8 encryption, uint32 timeout) {

	HTTPReply* reply = NULL;

	uint64 conid, location;
	NetworkChannel* channel;
	
	if (encryption == SSLENC)
		channel = createTCPConnection(addr, port, SSLENC, PROTOCOL_HTTP_CLIENT, false, false, 0, NULL, conid, location);
	else
		channel = createTCPConnection(addr, port, NOENC, PROTOCOL_HTTP_CLIENT, false, false, 0, NULL, conid, location);

	if (!channel)
		return HTTPReply::CreateErrorReply(HTTP_SERVER_UNAVAILABLE);

	reply = channel->sendReceiveHTTPRequest(req, conid, timeout);
	channel->endConnection(conid);
	return reply;
}


/////////////////////////////////////////////////////////////
// Connections
/////////////////////////////////////////////////////////////

NetworkChannel::NetworkChannel(NetworkManager* manager) {
	receiver = NULL;
	this->manager = manager;
}

NetworkChannel::~NetworkChannel() {
	shutdown();
	manager = NULL;
}

bool NetworkChannel::isConnected(uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if (thread == NULL)
		return false;
	bool res = ( thread->con && thread->con->isConnected() );
	return res;
}

bool NetworkChannel::shutdown() {
	NetworkThread* thread;
	uint32 timeleft;
	std::map<uint16, NetworkThread*>::iterator it, itEnd;
	for (it = listeners.begin(), itEnd = listeners.end(); it != itEnd; ++it) {
		if ( (thread = it->second) != NULL) {
			thread->shouldContinue = false;
			timeleft = 100;
			while (thread->isRunning) {
				utils::Sleep(5);
				if ( (timeleft -= 5) <= 0)
					break;
			}
			if (thread->isRunning) {
				ThreadManager::TerminateThread(thread->threadID);
				thread->isRunning = false;
			}
			delete(thread);
		}
	}
	listeners.clear();

	std::map<uint64, NetworkThread*>::iterator it2, it2End;
	for (it2 = connectionThreads.begin(), it2End = connectionThreads.end(); it2 != it2End; ++it2) {
		if ( (thread = it2->second) != NULL) {
			thread->autoreconnect = false;
			thread->shouldContinue = false;
			timeleft = 200;
			while (thread->isRunning) {
				utils::Sleep(5);
				if ( (timeleft -= 5) <= 0)
					break;
			}
			if (thread->isRunning) {
				utils::Sleep(50);
				ThreadManager::TerminateThread(thread->threadID);
				utils::Sleep(50);
				thread->isRunning = false;
			}
			delete(thread);
		}
	}
	connectionThreads.clear();

	queueHTTPRequestsMutex.enter();
	while (!queueHTTPRequests.empty()) {
		delete(queueHTTPRequests.front());
		queueHTTPRequests.pop();
	}
	queueHTTPRequestsMutex.leave();

	queueHTTPRepliesMutex.enter();
	while (!queueHTTPReplies.empty()) {
		delete(queueHTTPReplies.front());
		queueHTTPReplies.pop();
	}
	queueHTTPRepliesMutex.leave();

	queueMessagesMutex.enter();
	while (!queueMessages.empty()) {
		delete(queueMessages.front());
		queueMessages.pop();
	}
	queueMessagesMutex.leave();

	queueTelnetLinesMutex.enter();
	while (!queueTelnetLines.empty()) {
		delete(queueTelnetLines.front());
		queueTelnetLines.pop();
	}
	queueTelnetLinesMutex.leave();

	eventQueueMutex.enter();
	while (!eventQueue.empty()) {
		delete(eventQueue.front());
		eventQueue.pop();
	}
	eventQueueMutex.leave();

	return true;
}

bool NetworkChannel::startListener(uint64 cid, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, uint32 protocolTimeout, bool isDefaultProtocol) {
	NetworkThread* thread = listeners[port];
	bool created = false;
	if (thread == NULL) {
		uint64 conid = cid;
		if (!conid)
			conid = manager->addConnection(this);
		thread = new NetworkThread(this, conid);
		created = true;
		// Try binding to the port
		thread->port = port;
		thread->listener = new TCPListener();
		if (manager->sslCertPath.size())
			thread->listener->setSSLCertificate(manager->sslCertPath.c_str(), manager->sslKeyPath.c_str());
		thread->isAsync = isAsync;
		if (!thread->listener->init(port, encryption)) {
			if (created)
				delete(thread);
			return false;
		}

		if (isDefaultProtocol)
			thread->defaultProtocol = protocol;
		thread->autoProtocolTimeout = protocolTimeout;

		if ( isDefaultProtocol && (protocolTimeout == 0) )
			thread->autoProtocols = 0;
		else
			thread->autoProtocols |= protocol;

		// Start the thread
		if (!ThreadManager::CreateThread(NetworkChannel::NetworkListenerRun, thread, thread->threadID)) {
			if (created)
				delete(thread);
			return false;
		}
		
		listeners[port] = thread;
	}
	else {
		// Already running, just add the protocol
		thread->isAsync = isAsync;
		if (isDefaultProtocol)
			thread->defaultProtocol = protocol;
		thread->autoProtocolTimeout = protocolTimeout;

		if ( isDefaultProtocol && (protocolTimeout == 0) )
			thread->autoProtocols = 0;
		else
			thread->autoProtocols |= protocol;
	}

	return true;
}

bool NetworkChannel::stopListener(uint16 port, uint8 protocol) {
	NetworkThread* thread = listeners[port];
	if (thread == NULL)
		return true;
	thread->autoProtocols &= ~protocol;
	if (thread->defaultProtocol == protocol)
		thread->defaultProtocol = 0;
	if ((thread->autoProtocols == 0) && (thread->defaultProtocol == 0)) {
		if (thread->isRunning) {
			thread->shouldContinue = false;
			utils::Sleep(100);
			ThreadManager::TerminateThread(thread->threadID);
			thread->isRunning = false;
		}
		listeners.erase(port);
		delete(thread);
	}
	return true;
}

uint64 NetworkChannel::createTCPConnection(const char* addr, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint64& location) {
	NetworkConnection* con;
	if (encryption == NOENC) {
		TCPConnection* tcpCon = new TCPConnection();
		if (!tcpCon->connect(addr, port, location)) {
			delete(tcpCon);
			return false;
		}
		con = tcpCon;
	}
	else if (encryption == SSLENC) {
		SSLConnection* sslCon = new SSLConnection();
		if (!sslCon->init()) {
			delete(sslCon);
			return false;
		}
		if (!sslCon->connect(addr, port, location)) {
			delete(sslCon);
			return false;
		}
		con = sslCon;
	}
	return startConnection(con, protocol, isAsync, autoreconnect);
}

uint64 NetworkChannel::createTCPConnection(uint64 location, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect) {
	NetworkConnection* con;
	if (encryption == NOENC) {
		TCPConnection* tcpCon = new TCPConnection();
		if (!tcpCon->connect(location)) {
			delete(tcpCon);
			return false;
		}
		con = tcpCon;
	}
	else if (encryption == SSLENC) {
		SSLConnection* sslCon = new SSLConnection();
		if (!sslCon->connect(location)) {
			delete(sslCon);
			return false;
		}
		con = sslCon;
	}
	return startConnection(con, protocol, isAsync, autoreconnect);
}

uint64 NetworkChannel::createTCPConnection(const uint32* addresses, uint16 addressCount, uint16 port, uint8 encryption, uint8 protocol, bool isAsync, bool autoreconnect, uint64& location) {
	NetworkConnection* con;
	if (encryption == NOENC) {
		TCPConnection* tcpCon = new TCPConnection();
		if (!tcpCon->connect(addresses, addressCount, port, location)) {
			delete(tcpCon);
			return false;
		}
		con = tcpCon;
	}
	else if (encryption == SSLENC) {
		SSLConnection* sslCon = new SSLConnection();
		if (!sslCon->connect(addresses, addressCount, port, location)) {
			delete(sslCon);
			return false;
		}
		con = sslCon;
	}
	return startConnection(con, protocol, isAsync, autoreconnect);
}

uint64 NetworkChannel::createUDPConnection(uint16 port, uint8 protocol, bool isAsync, bool autoreconnect) {
	UDPConnection* con = new UDPConnection();
	if (!con->connect(port)) {
		delete(con);
		return false;
	}
	uint64 conid = startConnection(con, protocol, isAsync, autoreconnect);
	if (!conid)
		return false;

	NetworkThread* thread = connectionThreads[conid];
	if (thread == NULL)
		return false;
	udpListeners[port] = thread;
	return true;
}

bool NetworkChannel::endUDPConnection(uint16 port) {
	NetworkThread* thread = udpListeners[port];
	if (!thread)
		return false;
	if (!endConnection(thread->id))
		return false;
	udpListeners.erase(port);
	return true;
}

uint8 NetworkChannel::getConnectionType(uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if (!thread || !thread->con)
		return 0;
	return thread->con->getConnectionType();
}

uint64 NetworkChannel::getRemoteAddress(uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if (!thread || !thread->con)
		return 0;
	return thread->con->getRemoteAddress();
}

bool NetworkChannel::endConnection(uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if (thread == NULL)
		return false;
	thread->autoreconnect = false;
	thread->shouldContinue = false;
	if (thread->isRunning) {
		uint32 timeleft = 200;
		while (thread->isRunning) {
			utils::Sleep(5);
			if ( (timeleft -= 5) <= 0)
				break;
		}
		if (thread->isRunning) {
			utils::Sleep(50);
			ThreadManager::TerminateThread(thread->threadID);
			utils::Sleep(50);
			thread->isRunning = false;
		}
	}
	delete(thread);
	connectionThreads.erase(conid);
	return true;
}

bool NetworkChannel::setNewReceiver(NetworkReceiver* recv) {
	receiver = recv;
	return true;
}


NetworkEvent* NetworkChannel::waitForNetworkEvent(uint32 ms) {
	uint64 start = GetTimeNow();
	int32 spent = 0;

	while (eventQueue.empty()) {
		eventQueueSemaphore.wait((int32)ms-spent);
		if ( eventQueue.empty() && ( (spent = GetTimeAgeMS(start)) >= (int32)ms) )
			return NULL;
	}

	eventQueueMutex.enter();
	NetworkEvent* e = eventQueue.front();
	eventQueue.pop();
	eventQueueMutex.leave();
	return e;
}

HTTPRequest* NetworkChannel::waitForHTTPRequest(uint64& conid, uint32 ms) {
	uint64 start = GetTimeNow();
	int32 spent = 0;

	while (queueHTTPRequests.empty()) {
		queueHTTPRequestsSemaphore.wait((int32)ms-spent);
		if ( queueHTTPRequests.empty() && ( (spent = GetTimeAgeMS(start)) >= (int32)ms) )
			return NULL;
	}

	queueHTTPRequestsMutex.enter();
	HTTPRequest* req = queueHTTPRequests.front();
	queueHTTPRequests.pop();
	queueHTTPRequestsMutex.leave();
	return req;
}

TelnetLine* NetworkChannel::waitForTelnetLine(uint64& conid, uint32 ms) {
	uint64 start = GetTimeNow();
	int32 spent = 0;

	while (queueTelnetLines.empty()) {
		queueTelnetLinesSemaphore.wait((int32)ms-spent);
		if ( queueTelnetLines.empty() && ( (spent = GetTimeAgeMS(start)) >= (int32)ms) )
			return NULL;
	}

	queueTelnetLinesMutex.enter();
	TelnetLine* line = queueTelnetLines.front();
	queueTelnetLines.pop();
	queueTelnetLinesMutex.leave();
	return line;
}

DataMessage* NetworkChannel::waitForMessage(uint64& conid, uint32 ms) {
	uint64 start = GetTimeNow();
	int32 spent = 0;

	while (queueMessages.empty()) {
		queueMessagesSemaphore.wait((int32)ms-spent);
		if ( queueMessages.empty() && ( (spent = GetTimeAgeMS(start)) >= (int32)ms) )
			return NULL;
	}

	// LogPrint(0,0,0,"Dequeue!!!");
	queueMessagesMutex.enter();
	DataMessage* msg = queueMessages.front();
	queueMessages.pop();
	queueMessagesMutex.leave();

	return msg;
}

HTTPReply* NetworkChannel::waitForHTTPReply(uint64& conid, uint32 ms) {
	uint64 start = GetTimeNow();
	int32 spent = 0;

	while (queueHTTPReplies.empty()) {
		queueHTTPRepliesSemaphore.wait((int32)ms-spent);
		if ( queueHTTPReplies.empty() && ( (spent = GetTimeAgeMS(start)) >= (int32)ms) )
			return NULL;
	}

	queueHTTPRepliesMutex.enter();
	HTTPReply* reply = queueHTTPReplies.front();
	queueHTTPReplies.pop();
	queueHTTPRepliesMutex.leave();
	return reply;
}



bool NetworkChannel::sendHTTPReply(HTTPReply* reply, uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return false;
	bool res = HTTPProtocol::SendHTTPReply(thread->con, reply);
	return res;
}

bool NetworkChannel::sendTelnetLine(TelnetLine* line, uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return false;
	bool res = TelnetProtocol::SendTelnetLine(thread->con, line);
	return res;
}

TelnetLine* NetworkChannel::sendReceiveTelnetLine(TelnetLine* line, uint64 conid, uint32 timeout, uint32 size) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return NULL;

	if (!thread->isAsync) {
		// clear buffer before sending
		thread->con->clearBuffer();
	}

	if (!TelnetProtocol::SendTelnetLine(thread->con, line))
		return NULL;
	
	if (size)
		return TelnetProtocol::ReceiveTelnetLine(thread->con, timeout, size);
	else
		return TelnetProtocol::ReceiveTelnetLine(thread->con, timeout);
}

bool NetworkChannel::sendMessage(DataMessage* msg, uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return false;
	bool res = MessageProtocol::SendMessage(thread->con, msg);
	if (!res)
		int a = 1;
	return res;
}

HTTPReply* NetworkChannel::sendReceiveHTTPRequest(HTTPRequest* req, uint64 conid, uint32 timeout) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return HTTPReply::CreateErrorReply(HTTP_SERVER_UNAVAILABLE);

	if (!HTTPProtocol::SendHTTPRequest(thread->con, req))
		return HTTPReply::CreateErrorReply(HTTP_SERVER_UNAVAILABLE);
	
	return HTTPProtocol::ReceiveHTTPReply(thread->con, timeout);
}

uint32 NetworkChannel::getOutputSpeed(uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return 0;
	return thread->con->getOutputSpeed();
}

uint32 NetworkChannel::getInputSpeed(uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return 0;
	return thread->con->getInputSpeed();
}

bool NetworkChannel::sendHTTPRequest(HTTPRequest* req, uint64 conid) {
	NetworkThread* thread = connectionThreads[conid];
	if ((thread == NULL) || (thread->con == NULL))
		return false;

	if (HTTPProtocol::SendHTTPRequest(thread->con, req)) {
	//	if (thread->lastRequest != NULL)
	//		delete(thread->lastRequest);
	//	thread->lastRequest = new HTTPRequest(req);
		return true;
	}
	else {
		return false;
	}
}

bool NetworkChannel::enterHTTPRequest(HTTPRequest* req, uint64 conid) {
	if (!receiver || !receiver->receiveHTTPRequest(req, this, conid)) {
		queueHTTPRequestsMutex.enter();
		queueHTTPRequests.push(req);
		queueHTTPRequestsSemaphore.signal();
		queueHTTPRequestsMutex.leave();
	}
	return true;
}

bool NetworkChannel::enterHTTPReply(HTTPReply* reply, HTTPRequest* req, uint64 conid) {
	if (!receiver || !receiver->receiveHTTPReply(reply, req, this, conid)) {
		queueHTTPRepliesMutex.enter();
		queueHTTPReplies.push(reply);
		queueHTTPRepliesSemaphore.signal();
		queueHTTPRepliesMutex.leave();
	}
	return true;
}

bool NetworkChannel::enterMessage(DataMessage* msg, uint64 conid) {
	if (!receiver || !receiver->receiveMessage(msg, this, conid)) {
		queueMessagesMutex.enter();
		queueMessages.push(msg);
		// LogPrint(0,0,0,"Queue size is now: %u (%u)", (uint32)queueMessages.size(), msg->getType()[15]);
		queueMessagesSemaphore.signal();
		queueMessagesMutex.leave();
	}
	return true;
}

bool NetworkChannel::enterTelnetLine(TelnetLine* line, uint64 conid) {
	if (!receiver || !receiver->receiveTelnetLine(line, this, conid)) {
		queueTelnetLinesMutex.enter();
		queueTelnetLines.push(line);
		queueTelnetLinesSemaphore.signal();
		queueTelnetLinesMutex.leave();
	}
	return true;
}


bool NetworkChannel::enterNetworkEvent(uint8 type, uint8 protocol, uint64 conid) {
	NetworkEvent* ev = new NetworkEvent;
	ev->cid = this->cid;
	ev->conid = conid;
	ev->time = GetTimeNow();
	ev->type = type;
	ev->protocol = protocol;
	if (!receiver || !receiver->receiveNetworkEvent(ev, this, conid)) {
		eventQueueMutex.enter();
		eventQueue.push(ev);
		eventQueueSemaphore.signal();
		eventQueueMutex.leave();
	}
	return true;
}

uint64 NetworkChannel::autoDetectConnection(NetworkConnection* con, uint16 port, uint32 autoProtocols, uint32 autoProtocolTimeout, uint32 defaultProtocol, bool isAsync, bool autoreconnect) {

	uint64 conid = manager->addConnection(this);
	if (!conid)
		return 0;
	NetworkThread* thread = new NetworkThread(this, conid);
	thread->defaultProtocol = defaultProtocol;
	thread->autoProtocols = autoProtocols;
	thread->autoProtocolTimeout = autoProtocolTimeout;
	thread->autoreconnect = autoreconnect;
	thread->isAsync = isAsync;
	thread->parent = this;
	// Try connecting...
	thread->con = con;

	connectionThreads[conid] = thread;

	// Start the thread
	if (!ThreadManager::CreateThread(ConnectionAutodetectRun, thread, thread->threadID)) {
		connectionThreads.erase(conid);
		delete(thread);
		return 0;
	}

	return conid;
}

uint64 NetworkChannel::startConnection(NetworkConnection* con, uint8 protocol, bool isAsync, bool autoreconnect) {
	uint64 conid = manager->addConnection(this);
	NetworkThread* thread = new NetworkThread(this, conid);
	thread->defaultProtocol = protocol;
	thread->autoreconnect = autoreconnect;
	thread->isAsync = isAsync;
	// Try connecting...
	thread->con = con;

	THREAD_FUNCTION func = NULL;
	switch(protocol) {
	case PROTOCOL_HTTP_SERVER:
		func = HTTPServerRun;
		break;
	case PROTOCOL_HTTP_CLIENT:
		func = HTTPClientRun;
		break;
	case PROTOCOL_MESSAGE:
		func = MessageConnectionRun;
		break;
	case PROTOCOL_TELNET:
		func = TelnetServerRun;
		break;
	default:
		break;
	}

	connectionThreads[conid] = thread;
	// Start the thread
	if (func) {
		if (!ThreadManager::CreateThread(func, thread, thread->threadID)) {
			connectionThreads.erase(conid);
			delete(thread);
			return 0;
		}
	}

	return conid;
}

THREAD_RET THREAD_FUNCTION_CALL NetworkChannel::NetworkListenerRun(THREAD_ARG arg) {

	NetworkThread* thread = (NetworkThread*) arg;
	if ((thread == NULL) || (thread->listener == NULL))
		thread_ret_val(1);
	thread->isRunning = true;

	uint64 conID;
	NetworkConnection* con;
	while (thread->shouldContinue) {
		if ( (con = thread->listener->acceptConnection(50)) != NULL) {
			conID = thread->parent->autoDetectConnection(con, thread->port, thread->autoProtocols, thread->autoProtocolTimeout, thread->defaultProtocol, thread->isAsync, false);
			if (conID == 0) {
				delete(con);
				con = NULL;
			}
		}
	}

	thread->isRunning = false;
	thread_ret_val(0);
}


THREAD_RET THREAD_FUNCTION_CALL NetworkChannel::ConnectionAutodetectRun(THREAD_ARG arg) {

	NetworkThread* thread = (NetworkThread*) arg;
	if ((thread == NULL) || (thread->con == NULL))
		thread_ret_val(1);

	thread->isRunning = true;

	uint8 protocol = 0;
	uint32 maxSize = 1024;
	char* buffer = new char[maxSize];
	uint32 size = 0;

	uint64 start = GetTimeNow();

	if (thread->autoProtocols > 0) {
	//	printf("Autodetecting protocol");
		do {
			//printf(".");
			if (!thread->con->receiveAvailable(buffer, size, maxSize, 10, true)) {
				delete(thread->con);
				thread->con = NULL;
				thread->isRunning = false;
				delete [] buffer;
				thread_ret_val(1);
			}
			if (size > 0) {
				// utils::PrintBinary(buffer, size, false, "Autodetection");
				//	printf(" [%u]", size);

				if ( (thread->autoProtocols & PROTOCOL_HTTP_SERVER)
					&& HTTPProtocol::CheckBufferForCompatibility(buffer, size) ) {
						protocol = PROTOCOL_HTTP_SERVER;
						// printf("Autodetected HTTP protocol after %u ms...\n\n", GetTimeAgeMS(start));
				}
				else if ( (thread->autoProtocols & PROTOCOL_MESSAGE)
					&& MessageProtocol::CheckBufferForCompatibility(buffer, size) ) {
						protocol = PROTOCOL_MESSAGE;
						// printf("Autodetected Message protocol after %u ms...\n\n", GetTimeAgeMS(start));
				}
				else if ( (thread->autoProtocols & PROTOCOL_TELNET)
					&& TelnetProtocol::CheckBufferForCompatibility(buffer, size) ) {
						protocol = PROTOCOL_TELNET;
						// printf("Autodetected Telnet protocol after %u ms...\n\n", GetTimeAgeMS(start));
				}
			}
		//	else
		//		printf(".", size);
		} while (thread->con && (!protocol) && (GetTimeAgeMS(start) < (int32)thread->autoProtocolTimeout) && thread->shouldContinue);
		//printf("\n\n");
	}
	
	delete [] buffer;

	if (!thread->shouldContinue) {
		delete(thread->con);
		thread->con = NULL;
		thread->isRunning = false;
		thread_ret_val(1);
	}

	if (protocol == 0) {
		if (thread->defaultProtocol == 0) {
			LogPrint(0,LOG_NETWORK,2,"No valid protocol detected for incoming network connection, disconnecting...\n\n");
			// con->disconnect();
			delete(thread->con);
			thread->con = NULL;
			thread->isRunning = false;
			thread_ret_val(1);
		}
		else {
			protocol = thread->defaultProtocol;
			switch(protocol) {
			case PROTOCOL_HTTP_SERVER:
				//printf("Choosing default HTTP SERVER protocol after %u ms...\n\n", GetTimeAgeMS(start));
				break;
			case PROTOCOL_HTTP_CLIENT:
				//printf("Choosing default HTTP CLIENT protocol after %u ms...\n\n", GetTimeAgeMS(start));
				break;
			case PROTOCOL_MESSAGE:
				//printf("Choosing default MESSAGE protocol after %u ms...\n\n", GetTimeAgeMS(start));
				break;
			case PROTOCOL_TELNET:
				//printf("Choosing default TELNET protocol after %u ms...\n\n", GetTimeAgeMS(start));
				break;
			default:
				//printf("Choosing default unknown protocol after %u ms...\n\n", GetTimeAgeMS(start));
				break;
			}
		}
	}

	thread->parent->enterNetworkEvent(NETWORKEVENT_CONNECT, protocol, thread->id);

	thread->defaultProtocol = protocol;

	THREAD_FUNCTION func = NULL;
	switch(protocol) {
	case PROTOCOL_HTTP_SERVER:
		return HTTPServerRun(thread);
	case PROTOCOL_HTTP_CLIENT:
		return HTTPClientRun(thread);
	case PROTOCOL_MESSAGE:
		return MessageConnectionRun(thread);
	case PROTOCOL_TELNET:
		return TelnetServerRun(thread);
	default:
		delete(thread->con);
		thread->con = NULL;
		thread->isRunning = false;
		thread_ret_val(1);
	}


}


THREAD_RET THREAD_FUNCTION_CALL NetworkChannel::HTTPClientRun(THREAD_ARG arg) {

	NetworkThread* thread = (NetworkThread*) arg;
	if ((thread == NULL) || (thread->con == NULL))
		thread_ret_val(1);
	thread->isRunning = true;

	bool disconnected = false;

	// The main job here is to check for disconnects and to auto-reconnect

	while (thread->shouldContinue) {
		if (thread->con->isConnected()) {
			utils::Sleep(50);
		}
		else if (thread->autoreconnect) {
			if (!disconnected) {
				thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT_RETRYING, thread->defaultProtocol, thread->id);
				disconnected = true;
			}
			if (!thread->con->reconnect()) {
				utils::Sleep(50);
			}
			else {
				thread->parent->enterNetworkEvent(NETWORKEVENT_RECONNECT, thread->defaultProtocol, thread->id);
				disconnected = false;
			}
		}
		else {
			thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT, thread->defaultProtocol, thread->id);
			thread->isRunning = false;
			thread_ret_val(1);
		}
	}

	//	printf("9");
	thread->isRunning = false;
	thread_ret_val(0);
}

THREAD_RET THREAD_FUNCTION_CALL NetworkChannel::HTTPServerRun(THREAD_ARG arg) {

	// printf("--- Starting new HTTP Server thread...\n");
	NetworkThread* thread = (NetworkThread*) arg;
	if ((thread == NULL) || (thread->con == NULL))
		thread_ret_val(1);
	thread->isRunning = true;

	bool disconnected = false;

	HTTPRequest* req;
	while (thread->shouldContinue) {
		if (thread->con->isConnected()) {
			// printf("--- HTTP Server thread receiving...\n");
			req = HTTPProtocol::ReceiveHTTPRequest(thread->con, 500);
			if (req) {
				// printf("--- HTTP Server thread got new request (%s)...\n", req->getRequest());
				if (!thread->parent->enterHTTPRequest(req, thread->id)) {
					thread->parent->enterNetworkEvent(NETWORKEVENT_UNPROCESSED_DATA, thread->defaultProtocol, thread->id);
					delete(req);
				}
			}
		}
		//else if (thread->autoreconnect) {
		//	if (!disconnected) {
		//		thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT_RETRYING, thread->defaultProtocol, thread->id);
		//		disconnected = true;
		//	}
		//	if (!thread->con->reconnect()) {
		//		utils::Sleep(50);
		//	}
		//	else {
		//		thread->parent->enterNetworkEvent(NETWORKEVENT_RECONNECT, thread->defaultProtocol, thread->id);
		//		disconnected = false;
		//	}
		//}
		else {
			thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT, thread->defaultProtocol, thread->id);
			thread->isRunning = false;
			// printf("--- Disconnect, HTTP Server thread exit...\n");
			thread_ret_val(1);
		}
	}

	thread->isRunning = false;
	// printf("--- Finish, HTTP Server thread exit...\n");
	thread_ret_val(0);
}

THREAD_RET THREAD_FUNCTION_CALL NetworkChannel::MessageConnectionRun(THREAD_ARG arg) {

	NetworkThread* thread = (NetworkThread*) arg;
	if ((thread == NULL) || (thread->con == NULL))
		thread_ret_val(1);
	thread->isRunning = true;

	bool disconnected = false;

	uint64 t;
	DataMessage* msg;
	while (thread->shouldContinue) {
		if (thread->con->isConnected()) {
			t = GetTimeNow();
			msg = MessageProtocol::ReceiveMessage(thread->con, 30);
			//if (GetTimeAge(t) > 200) {
				//if (msg)
				//	LogPrint(0,0,0,"************** ReceiveMessage msg (tag %u) took %s ******************", msg->getTag(), PrintTimeDifString(GetTimeAge(t)).c_str());
				//else
				//	LogPrint(0,0,0,"************** ReceiveMessage NULL took %s ******************", PrintTimeDifString(GetTimeAge(t)).c_str());
			//}
			if (msg) {
				//t = GetTimeNow();
				if (!thread->parent->enterMessage(msg, thread->id)) {
					thread->parent->enterNetworkEvent(NETWORKEVENT_UNPROCESSED_DATA, thread->defaultProtocol, thread->id);
					delete(msg);
				}
			}
		}
		else if (thread->autoreconnect) {
		//	LogPrint(0,0,0,"************** ReceiveMessage auto reconnect ******************");
			if (!disconnected) {
				thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT_RETRYING, thread->defaultProtocol, thread->id);
				disconnected = true;
			}
			if (!thread->con->reconnect()) {
				utils::Sleep(50);
			}
			else {
				utils::Sleep(20);
				if (thread->con->isConnected()) {
					thread->parent->enterNetworkEvent(NETWORKEVENT_RECONNECT, thread->defaultProtocol, thread->id);
					disconnected = false;
				}
				// otherwise, it didn't work anyway...
			}
		}
		else {
		//	LogPrint(0,0,0,"************** ReceiveMessage disconnect ******************");
			thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT, thread->defaultProtocol, thread->id);
			thread->isRunning = false;
			thread_ret_val(1);
		}
	}

//	LogPrint(0,0,0,"************** ReceiveMessage returning ******************");
	thread->isRunning = false;
	thread_ret_val(0);
}

THREAD_RET THREAD_FUNCTION_CALL NetworkChannel::TelnetServerRun(THREAD_ARG arg) {

	NetworkThread* thread = (NetworkThread*) arg;
	if ((thread == NULL) || (thread->con == NULL))
		thread_ret_val(1);
	thread->isRunning = true;

	bool disconnected = false;

	TelnetLine* line;
	while (thread->shouldContinue) {
		if (thread->con->isConnected(thread->isAsync ? 0 : 50)) {
			if (thread->isAsync) {
				line = TelnetProtocol::ReceiveTelnetLine(thread->con, 50);
				if (line) {
					if (!thread->parent->enterTelnetLine(line, thread->id)) {
						thread->parent->enterNetworkEvent(NETWORKEVENT_UNPROCESSED_DATA, thread->defaultProtocol, thread->id);
						delete(line);
					}
				}
			}
			else {
				utils::Sleep(50);
			}
			//else {
			//	if (thread->con->peekStream() < 0) {
			//		thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT, thread->defaultProtocol, thread->id);
			//		thread->isRunning = false;
			//		thread_ret_val(1);
			//	}
			//}
		}
		else {
			thread->parent->enterNetworkEvent(NETWORKEVENT_DISCONNECT, thread->defaultProtocol, thread->id);
			thread->isRunning = false;
			thread_ret_val(1);
		}
	}

	thread->isRunning = false;
	thread_ret_val(0);
}

bool NetworkManager::UnitTest() {

	char* dat;
	uint64 conid = 0;
	uint64 conid2 = 0;
	uint32 address;
	uint64 destination;
	DataMessage* msg, *msg2;
	uint64 start = GetTimeNow();
	uint32 count = 100, subcount = 100, subcount2 = 10, n, m, k, size;
	uint64 t, end;
	NetworkChannel* con = NULL;
	PsyType type = { { 1,10001,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };

	uint64 recvid;
	printf("Testing Network Manager...\n\n");

	NetworkManager* manager = new NetworkManager();

	NetworkChannel* listen;
		
	listen = manager->createListener(10000, NOENC, PROTOCOL_MESSAGE, true, 0, true, 0, NULL);
	if (!listen) {
		printf("Could not start listening...\n\n");
		goto err;
	}

	uint64 location;
	con = manager->createTCPConnection("localhost", 10000, NOENC, PROTOCOL_MESSAGE, true, false, 0, NULL, conid, location);
	if (!con) {
		printf("Could not connect...\n\n");
		goto err;
	}
//	printf("Connected...\n\n");
	size = 1024;
	dat = new char[size];
	memset(dat, 0, size);
	msg = new DataMessage(CTRL_TEST, 20);
	msg->setData("Test", dat, size);
	delete [] dat;
	if (!con->sendMessage(msg, conid)) {
		printf("Could not send message!\n\n");
		goto err;
	}
	msg2 = listen->waitForMessage(recvid, 1000);
	if (!msg2) {
		printf("No message received!\n\n");
		goto err;
	}
	else if ( (msg->getType() != msg2->getType()) || (msg->getFrom() != msg2->getFrom()) ) {
		printf("Message sent and received mismatch!\n\n");
		goto err;
	}
	delete(msg2);

//	printf("Sending message...\n\n");
	start = GetTimeNow();
	if (!con->sendMessage(msg, conid)) {
		printf("Could not send message!\n\n");
		goto err;
	}
	t = GetTimeNow();
	msg2 = listen->waitForMessage(recvid, 1000);

	if (!msg2) {
		printf("No message received!\n\n");
		goto err;
	}
	else if ( (msg->getType() != msg2->getType()) || (msg->getFrom() != msg2->getFrom()) ) {
		printf("Message sent and received mismatch!\n\n");
		goto err;
	}
	delete(msg2);
	end = GetTimeNow();
	printf("Single message send %s, receive %s, total %s, starting test...\n\n",
		PrintTimeDifString(t - start).c_str(),
		PrintTimeDifString(end - t).c_str(),
		PrintTimeDifString(end - start).c_str() );

	t = 0;
	count = 10, subcount = 10, subcount2 = 15;
	for (m = 0; m < count; m++) {
	//	LogPrint(0,0,0,"[%u] Starting to send %u msgs...\n", (m+1)*subcount, subcount);
		start = GetTimeNow();
		for (n = 0; n < subcount; n++) {
			type.levels[15] = n;
			msg->setType(type);
			// t = GetTimeNow();
			for (k=0; k<subcount2; k++) {
				if (!con->sendMessage(msg, conid)) {
					printf("Could not send message %u!\n\n", n);
					goto err;
				}
			}
		//	printf("SendMessage:      %lu\n", GetTimeAgeMS(t));
//		}
	//	LogPrint(0,0,0,"[%u] Sent %u msgs...\n", (m+1)*subcount, subcount);
		// utils::Sleep(1000);
		// printf("Msg Queue size: %u\n\n", (uint32)listen->queueMessages.size());
//		for (n = 0; n < subcount; n++) {
			for (k=0; k<subcount2; k++) {
				msg2 = listen->waitForMessage(recvid, 100);
				type.levels[15] = n;
				if (!msg2) {
					printf("[%u/%u/%u] No message received!\n\n", n, m, k);
					goto err;
				}
				else if ( (msg2->getType() != type) || (msg->getFrom() != msg2->getFrom()) ) {
					printf("[%u] Message sent and received mismatch!\n\n", n);
					goto err;
				}
				delete(msg2);
			}
		}
		end = GetTimeNow();
		LogPrint(0,LOG_NETWORK,1,"[%u/%u] Sent and received %u msgs (%u bytes), %.3fus per msg",
			(m+1)*subcount*subcount2, (m+1)*subcount*subcount2 * size,
			subcount*subcount2, subcount*subcount2 * size,
			((double)(end-start))/(subcount*subcount2));
		t += end - start;

		//listen->sendMessage(msg, 2);
		//msg2 = con->waitForMessage(conid, 1000);
		//delete(msg2);
	//	utils::Sleep(10);
	}

	LogPrint(0,LOG_NETWORK,1,"Total: Sent and received %u msgs, %.3fus per msg\n", count*subcount*subcount2, ((double)t)/(count*subcount*subcount2));

	con->endConnection(conid);
	delete(msg);






	t = 0;

	listen = manager->createUDPConnection(10001, PROTOCOL_MESSAGE, true, true, 0, NULL, conid2);
	if (!listen) {
		printf("Could not start UDP listening...\n\n");
		goto err;
	}

	utils::GetLocalIPAddress(address);
	destination = GETIPADDRESSPORT(address, 10001);

//	printf("Connected...\n\n");
	size = 1024;
	dat = new char[size];
	memset(dat, 0, size);
	msg = new DataMessage(CTRL_TEST, 20);
	msg->setData("Test", dat, size);
	delete [] dat;
	if (!manager->sendUDPMessage(msg, destination)) {
		printf("Could not send message!\n\n");
		goto err;
	}
	msg2 = listen->waitForMessage(recvid, 100000);
	if (!msg2) {
		printf("No message received!\n\n");
		goto err;
	}
	else if ( (msg->getType() != msg2->getType()) || (msg->getFrom() != msg2->getFrom()) ) {
		printf("Message sent and received mismatch!\n\n");
		goto err;
	}
	delete(msg2);

//	printf("Sending message...\n\n");
	start = GetTimeNow();
	if (!manager->sendUDPMessage(msg, destination)) {
		printf("Could not send message!\n\n");
		goto err;
	}
	t = GetTimeNow();
	msg2 = listen->waitForMessage(recvid, 1000);

	if (!msg2) {
		printf("No message received!\n\n");
		goto err;
	}
	else if ( (msg->getType() != msg2->getType()) || (msg->getFrom() != msg2->getFrom()) ) {
		printf("Message sent and received mismatch!\n\n");
		goto err;
	}
	delete(msg2);
	end = GetTimeNow();
	printf("Single message send %s, receive %s, total %s, starting test...\n\n",
		PrintTimeDifString(t - start).c_str(),
		PrintTimeDifString(end - t).c_str(),
		PrintTimeDifString(end - start).c_str() );

	t = 0;
	count = 50, subcount = 10, subcount2 = 1;
	for (m = 0; m < count; m++) {
	//	LogPrint(0,0,0,"[%u] Starting to send %u msgs...\n", (m+1)*subcount, subcount);
		start = GetTimeNow();
		for (n = 0; n < subcount; n++) {
			type.levels[15] = n;
			msg->setType(type);
			// t = GetTimeNow();
			for (k=0; k<subcount2; k++) {
				if (!manager->sendUDPMessage(msg, destination)) {
					printf("Could not send message %u!\n\n", n);
					goto err;
				}
			}
		//	printf("SendMessage:      %lu\n", GetTimeAgeMS(t));
//		}
	//	LogPrint(0,0,0,"[%u] Sent %u msgs...\n", (m+1)*subcount, subcount);
		// utils::Sleep(1000);
		// printf("Msg Queue size: %u\n\n", (uint32)listen->queueMessages.size());
//		for (n = 0; n < subcount; n++) {
			for (k=0; k<subcount2; k++) {
				msg2 = listen->waitForMessage(recvid, 100);
				type.levels[15] = n;
				if (!msg2) {
					printf("[%u/%u/%u] No message received!\n", n, m, k);
					t -= 100000;
					// goto err;
				}
				else if ( (msg2->getType() != type) || (msg->getFrom() != msg2->getFrom()) ) {
					printf("[%u/%u/%u] Message sent and received mismatch!\n", n, m, k);
					// goto err;
				}
				//printf("[%u] Message sent and received!\n\n", n);
				delete(msg2);
			}
		}
		end = GetTimeNow();
		LogPrint(0,LOG_NETWORK,1,"[%u/%u] Sent and received %u msgs (%u bytes), %.3fus per msg",
			(m+1)*subcount*subcount2, (m+1)*subcount*subcount2 * size,
			subcount*subcount2, subcount*subcount2 * size,
			((double)(end-start))/(subcount*subcount2));
		t += end - start;

		//listen->sendMessage(msg, 2);
		//msg2 = con->waitForMessage(conid, 1000);
		//delete(msg2);
	//	utils::Sleep(10);
	}

	delete(msg);
	LogPrint(0,LOG_NETWORK,1,"Total UDP: Sent and received %u msgs, %.3fus per msg\n", count*subcount*subcount2, ((double)t)/(count*subcount*subcount2));

	printf("Network Manager test success!\n\n");
	delete(manager);
	ThreadManager::Shutdown();
	return true;
err:
	printf("Network Manager test failed!\n\n");
	delete(manager);
	ThreadManager::Shutdown();
	return false;
}


HTTPTestServer::HTTPTestServer() {
}

HTTPTestServer::~HTTPTestServer() {
}

bool HTTPTestServer::receiveHTTPRequest(HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
//	printf("HTTPTestServer received HTTPRequest...\n\n");

	char text[512];
	snprintf(text, 512, "Hello World %llu", GetTimeNow());

	uint64 localAddr = 0;
	utils::GetLocalIPAddress(*(uint32*)&localAddr);

	HTTPReply* reply = new HTTPReply(localAddr);
	if (!reply->createPage(HTTP_OK, GetTimeNow(), "MyServer", GetTimeNow(), true, false, "text/html", text)) {
		printf("Error generating HTML page...\n\n");
	}

	if (!channel->sendHTTPReply(reply, conid))
		printf("Error sending response...\n\n");
//	else
//		printf("HTTPTestServer sent response...\n\n");

	delete(reply);
	return true;
}

bool NetworkManager::UnitTestHTTP() {

//	char buffer1[] = 
//"GET /path/file.html HTTP/1.0\n\
//From: someuser@jmarshall.com\n\
//User-Agent: HTTPTool/1.0\n\
//if-modified-since: Sat, 29 Oct 1994 19:43:31 GMT\n\n";
//
//	char buffer[] = 
//"POST /path/script.cgi HTTP/1.0\n\
//From: frog@jmarshall.com\n\
//User-Agent: HTTPTool/1.0\n\
//Content-Type: application/x-www-form-urlencoded\n\
//Content-Length: 35\n\n\
//home=Cosby&favorite+flavor=fl%26ies";
//
//	HTTPRequest* req = new HTTPRequest(0);
//	if (!req->processHeader(buffer, strlen(buffer)))
//		return false;
//
//	if (!req->processContent("home=Cosby&favorite+flavor=fl%26ies", 35))
//		return false;

	uint64 conid;
	NetworkChannel* con;
	HTTPRequest* req;
	HTTPReply* reply;
	uint32 n, count;

	printf("Testing Network Manager HTTP...\n\n");

	HTTPTestServer* testServer = new HTTPTestServer();
	NetworkManager* manager = new NetworkManager();

	NetworkChannel* listen = manager->createListener(10000, NOENC, PROTOCOL_HTTP_SERVER, true, 3000, false, 0, testServer);
	if (!listen) {
		printf("Could not start listening...\n\n");
		goto err;
	}

	utils::Sleep(1000);

	uint64 location;
	con = manager->createTCPConnection("localhost", 10000, NOENC, PROTOCOL_HTTP_CLIENT, true, false, 0, NULL, conid, location);
	if (!con) {
		printf("Could not connect...\n\n");
		goto err;
	}
	printf("Connected...\n\n");

	req = new HTTPRequest((uint64)0);
	if (!req->createRequest(HTTP_GET, "", "/", NULL, 0, true, 0)) {
		delete(req);
		printf("Could not create request...\n\n");
		goto err;
	}
	reply = con->sendReceiveHTTPRequest(req, conid, 3000);
	delete(req);
	if (reply == NULL) {
		printf("Did not receive reply...\n\n");
		goto err;
	}
	delete(reply);

	count = 1000;
	for (n=0; n<count; n++) {
		if (n && !(n%10)) {
			con->endConnection(conid);
			con = manager->createTCPConnection("localhost", 10000, NOENC, PROTOCOL_HTTP_CLIENT, true, false, 0, NULL, conid, location);
		}

		req = new HTTPRequest((uint64)0);
		if (!req->createRequest(HTTP_GET, "", "/", NULL, 0, true, 0)) {
			delete(req);
			printf("Could not create request [%u]...\n\n", n);
			goto err;
		}
		reply = con->sendReceiveHTTPRequest(req, conid, 3000);
		delete(req);
		if (reply == NULL) {
			printf("Did not receive reply [%u]...\n\n", n);
			goto err;
		}
		delete(reply);
	}

	con->endConnection(conid);


	//NetworkChannel* con2 = manager->createConnection("cmlabs.com", 80, PROTOCOL_HTTP_CLIENT, false, 0, NULL, conid);
	//if (!con2) {
	//	printf("Could not connect to cmlabs.com...\n\n");
	//	goto err;
	//}
	//printf("Connected...\n\n");

	//req = new HTTPRequest((uint64)0);
	//if (!req->createRequest(HTTP_GET, "cmlabs.com", "/", NULL, 0, true, 0)) {
	//	delete(req);
	//	printf("Could not create request...\n\n");
	//	goto err;
	//}
	//printf("Request:\n%s\n", req->data);
	//if (!con2->sendHTTPRequest(req, conid)) {
	//	delete(req);
	//	printf("Could not send request...\n\n");
	//	goto err;
	//}
	//delete(req);

	//reply = con2->waitForHTTPReply(conid2, 10000);
	//if (reply == NULL) {
	//	printf("Did not receive reply...\n\n");
	//	goto err;
	//}
	//printf("Reply:\n%s\n", reply->data);
	//delete(reply);






//	utils::Sleep(10000000);

	printf("Network Manager HTTP test success!\n\n");
	delete(manager);
	return true;
err:
	printf("Network Manager test failed!\n\n");
	delete(manager);
	return false;

}



struct HTTPServerTestData {
	const char* host;
	uint32 port;
	std::vector<std::string>* urls;
	NetworkManager* manager;
	uint32 status;
};

THREAD_RET THREAD_FUNCTION_CALL HTTPServerTest(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	
	HTTPServerTestData* data = (HTTPServerTestData*) arg;
	data->status = 1;

	uint64 conid;
	NetworkChannel* con = NULL;
	HTTPRequest* req;
	HTTPReply* reply;
	uint64 location;
	uint32 e;

	utils::SeedRandomValues();

	printf("Connecting...\n");
	con = data->manager->createTCPConnection(data->host, data->port, NOENC, PROTOCOL_HTTP_CLIENT, true, false, 0, NULL, conid, location);
	if (!con) {
		printf("Could not connect to host %s:%u...\n\n", data->host, data->port);
		data->status = 90;
		goto err;
	}
	data->status = 2;

	for (uint32 n=0; n<10; n++) {
		// choose a random url...
		e = (uint32)utils::RandomValue((double)(data->urls->size()-1));
		// printf("Getting %u url: %s...\n\n", e, data->urls->at(e).c_str());

		req = new HTTPRequest((uint64)0);
		if (!req->createRequest(HTTP_GET, data->host, data->urls->at(e).c_str(), NULL, 0, true, 0)) {
			delete(req);
			printf("Could not create request [%u]...\n\n", n);
			data->status = 98;
			goto err;
		}

		printf("[%u] Sending...\n", n);
		reply = con->sendReceiveHTTPRequest(req, conid, 3000);
		delete(req);

		if (reply == NULL) {
			printf("Did not receive reply...\n\n");
			data->status = 97;
			goto err;
		}
		else if (reply->type == HTTP_SERVER_UNAVAILABLE) {
			printf("Server unavailable...\n\n");
			data->status = 96;
			goto err;
		}
		else if (reply->type == HTTP_SERVER_NOREPLY) {
			printf("Server no reply...\n\n");
			data->status = 95;
			goto err;
		}
		else if (reply->type == HTTP_SERVER_MALFORMED_REPLY) {
			printf("Server malformed reply...\n\n");
			data->status = 94;
			goto err;
		}

		delete(reply);
	}

	data->status = 10;
	thread_ret_val(0);
err:
	thread_ret_val(1);
}

bool NetworkManager::TestHTTP(const char* host, uint32 port, std::vector<std::string> &urls) {

	printf("Testing HTTP on %s:%u with random urls...\n\n", host, port);

	uint32 loops = 10;
	uint32 numThreads = 5;
	uint64 conid;
	uint64 location;

	HTTPServerTestData* data = NULL;

	NetworkManager* manager = new NetworkManager();
	NetworkChannel* testChannel = manager->createTCPConnection(host, port, NOENC, PROTOCOL_HTTP_CLIENT, false, false, 0, NULL, conid, location);
	if (!testChannel) {
		printf("Could not connect to host %s:%u...\n\n", host, port);
		delete(manager);
		return false;
	}
	testChannel->endConnection(conid);
	printf("Host is available, starting test...\n\n");

	//HTTPRequest* req;
	//HTTPReply* reply;

	//req = new HTTPRequest((uint64)0);
	//if (!req->createRequest(HTTP_GET, host, "/", NULL, 0, true, 0)) {
	//	delete(req);
	//	printf("Could not create request...\n\n");
	//	delete(manager);
	//	return false;
	//}

	//reply = testChannel->sendReceiveHTTPRequest(req, conid, 3000);
	//delete(req);

	//if (reply == NULL) {
	//	printf("Did not receive reply...\n\n");
	//	delete(manager);
	//	return false;
	//}
	//delete(reply);

//	for (uint32 n=0; n<100; n++) {
//		for (uint32 m=0; m<100; m++) {
//			// choose a random url...
//			uint32 e = (uint32)utils::RandomValue(urls.size()-1);
//			printf("Getting url: %s... ", urls.at(e).c_str());
//
//			req = new HTTPRequest((uint64)0);
//			if (!req->createRequest(HTTP_GET, host, urls.at(e).c_str(), NULL, 0, true, 0)) {
//				delete(req);
//				printf("Could not create request [%u]...\n\n", n);
//				goto err;
//			}
//
//			reply = testChannel->sendReceiveHTTPRequest(req, conid, 3000);
//			delete(req);
//
//			if (reply == NULL) {
//				printf("Did not receive reply...\n\n");
//				goto err;
//			}
//			printf("SUCCESS\n");
//			delete(reply);
//		}
//		testChannel->endConnection(conid);
//		testChannel = manager->createTCPConnection(host, port, PROTOCOL_HTTP_CLIENT, false, false, 0, NULL, conid, location);
//		if (!testChannel) {
//			printf("Could not connect %u to host %s:%u...\n\n", n, host, port);
//			delete(manager);
//			return false;
//		}
//	}
//
//
//err:
//	delete(manager);
//	return false;

	//testChannel->endConnection(conid);
	//printf("Host is available, starting test...\n\n");

	uint32 n, m;
	uint32 threadIDs[1000];
	HTTPServerTestData testData[1000];
	for (n=0; n<loops; n++) {
		for (m=0; m<numThreads; m++) {
			testData[m].host = host;
			testData[m].port = port;
			testData[m].urls = &urls;
			testData[m].manager = manager;
			testData[m].status = 0;
			if (!ThreadManager::CreateThread(HTTPServerTest, &testData[m], threadIDs[m])) {
				LogPrint(0, LOG_SYSTEM, 0, "Could not start tester thread...");
				return false;
			}
		}
		while (true) {
			for (m=0; m<numThreads; m++) {
				if (ThreadManager::IsThreadRunning(threadIDs[m]))
					break;
				if (testData[m].status > 10) {
					printf("[%u] Failed\n", m);
					return false;
				}
				else {
					printf("[%u] Done\n", m);
				}
			}
			if (m >= numThreads)
				break;
			utils::Sleep(100);
		}
	}

	return true;
}


}


