

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of CMSDK version 2.1.1.260726 (26 July 2026)
//_/_/  
//_/_/  developed by
//_/_/  
//_/_/  Communicative Machines Ltd. 
//_/_/    http://cmlabs.com/
//_/_/  Copyright (c) 2026
//_/_/  
//_/_/  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.
//_/_/
//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
/*! @file NetworkConnections.cpp
	@brief Implementation of the raw socket transports: TCPListener,
	TCPConnection, UDPConnection and SSLConnection (OpenSSL). Contains the
	platform-specific socket code — Winsock (WSAStartup, SOCKET) under
	Windows vs BSD sockets under POSIX — buffered receive logic and the SSL
	peer/hostname verification policy.
	@ingroup Networking */
#include "NetworkConnections.h"
#ifndef WINDOWS
	#include <arpa/inet.h>
#endif // WINDOWS

namespace	cmlabs{

#ifdef _USE_SSL_
// Drain OpenSSL's thread-local error queue and route it through the CMSDK log
// system instead of dumping it straight to stdout/stderr with
// ERR_print_errors_fp(). Two reasons: (1) it keeps SSL diagnostics on the
// normal LOG_NETWORK channel (respecting log level and receiver), so the unit
// test harness - which silences the log in non-verbose mode - no longer shows
// raw OpenSSL lines from the negative-path verify tests; (2) real failures are
// still visible when network logging is enabled. @param level LogPrint level.
static void LogSSLErrors(const char* context, int level) {
	unsigned long e;
	bool any = false;
	char buf[256];
	while ((e = ERR_get_error()) != 0) {
		ERR_error_string_n(e, buf, sizeof(buf));
		LogPrint(0, LOG_NETWORK, level, "SSL: %s: %s", context ? context : "error", buf);
		any = true;
	}
	if (!any)
		LogPrint(0, LOG_NETWORK, level, "SSL: %s (no OpenSSL error detail)", context ? context : "error");
}
#endif // _USE_SSL_

/////////////////////////////////////////////////////////////
// SSL Encryption Protocol
/////////////////////////////////////////////////////////////

bool SSLCheckBufferForCompatibility(const char* buffer, uint32 length) {
	if (length < 14)
		return false;
	return ( utils::stristr(buffer, "ClientVersion ") == buffer );
}

bool AESCheckBufferForCompatibility(const char* buffer, uint32 length) {
	if (length < 14)
		return false;
	return ( utils::stristr(buffer, "xxxxxxxxx ") == buffer );
}


TCPListener::TCPListener() {
	CHECKNETWORKINIT

	encryption = NOENC;
	localAddress = 0;
	threadID = 0;
	lastActivity = 0;
	socket = INVALID_SOCKET;
	receiver = NULL;
	dataReceiver = NULL;
}

TCPListener::~TCPListener() {
	disconnect();
	localAddress = 0;
	threadID = 0;
	lastActivity = 0;
	socket = INVALID_SOCKET;
	receiver = NULL;
	dataReceiver = NULL;
}


void TCPListener::disconnectInternal(uint16 error) {
	stop();
	if (socket != INVALID_SOCKET) {
		shutdown(socket, SD_BOTH);
		closesocket(socket);
		socket = INVALID_SOCKET;
	}
	if (receiver && error)
		receiver->registerError(error, this);
}

bool TCPListener::disconnect(uint16 error) {
	mutex.enter(1000);
	disconnectInternal(error);
	mutex.leave();
	return true;
}

bool TCPListener::isConnected() {
	return (socket != INVALID_SOCKET);
}

bool TCPListener::init(uint16 port, uint8 encryption, NetworkConnectionReceiver* receiver, NetworkDataReceiver* dataReceiver) {
	if (port == 0) {
		LogPrint(0, LOG_NETWORK, 0, "Could not start TCPListener on port 0...");
		return false;
	}

	mutex.enter(1000);
	this->encryption = encryption;
	memcpy(((char*)&localAddress)+sizeof(uint32), &port, sizeof(uint16));
	// Open TCP port

	// Setup listening on port node->networkPort
	if((socket=::socket(AF_INET,SOCK_STREAM,IPPROTO_TCP))==INVALID_SOCKET){
		LogPrint(0, LOG_NETWORK, 0, "Could not create TCPListener socket (%d)...", utils::GetLastOSErrorNumber());
		mutex.leave();
		return false;
	}
  
	#ifdef WINDOWS
		// Set the exclusive address option, preventing other software binding to
		// non-INADDR_ANY (i.e. interface addresses such as localhost directly)
		int one = 1;
		setsockopt(socket, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, (char *) &one, sizeof(one));
	#else
		/*
			This socket option tells the kernel that even if this port is busy (in
			the TIME_WAIT state), go ahead and reuse it anyway.  If it is busy,
			but with another state, you will still get an address already in use
			error.  It is useful if your server has been shut down, and then
			restarted right away while sockets are still active on its port.  You
			should be aware that if any unexpected data comes in, it may confuse
			your server, but while this is possible, it is not likely.
		*/
		int one = 1;
		setsockopt(socket,SOL_SOCKET,SO_REUSEADDR,&one,sizeof(one));
	#endif

	struct	sockaddr_in	addr;
	addr.sin_family= AF_INET;
	addr.sin_addr.s_addr=INADDR_ANY;
	addr.sin_port=htons(port);

	if(bind(socket,(SOCKADDR*)&addr,sizeof(struct sockaddr_in))==SOCKET_ERROR){
		LogPrint(0, LOG_NETWORK, 0, "Could not create TCPListener on port %u...", port);
		disconnectInternal(0);
		mutex.leave();
		return false;
	}

	// Set blocking mode
	utils::SetSocketNonBlockingMode(socket);

	if(listen(socket,SOMAXCONN)==SOCKET_ERROR){
		disconnectInternal(0);
		mutex.leave();
		return false;
	}

	if (receiver != NULL) {
		this->receiver = receiver;
		// Start networking thread
		if (!ThreadManager::CreateThread(TCPListenerRun, this, threadID)) {
			LogPrint(0, LOG_NETWORK, 0, "Could not start TCPListener thread...");
			disconnectInternal(0);
			mutex.leave();
			return false;
		}
	}
	this->dataReceiver = dataReceiver;

	LogPrint(0, LOG_NETWORK, 3, "Started TCPListener thread on port %u...", port);
	mutex.leave();
	return true;
}

uint64 TCPListener::getLocalAddress() {
	return localAddress;
}

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

NetworkConnection* TCPListener::acceptConnection(uint32 timeout) {
	SOCKET new_sock;
	TCPConnection* newCon;
	#ifdef _USE_SSL_
		SSLConnection* sslCon;
	#endif // _USE_SSL_
	uint64 start = GetTimeNow(), timespent;

	while ( (timespent = (GetTimeNow() - start)/1000) < timeout) {
		// Listen for incoming connections
		mutex.enter(1000);
		new_sock = accept(socket, NULL, NULL);

		if ((int) new_sock < 0) {
			int err = utils::GetLastOSErrorNumber();
			mutex.leave();
			if ( (err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN) ) {
				// No connection available right now, continue waiting
				utils::WaitForSocketReadability(socket, (uint32)(timeout-timespent));
				// printf("[%u] ", GETIPPORT(localAddress));
			}
			else {
				LogPrint(0, LOG_NETWORK, 2, "Shutting down network listener with error code: %d", new_sock);
				// Socket error, disconnect and exit
				#ifdef WINDOWS
					if (err != WSAENOTSOCK)
				#endif
					disconnect(NETWORKERROR_ACCEPT);
				return NULL;
			}
		}
		else {
			mutex.leave();
			// We have a winner!
			// Check for encryption
			#ifdef _USE_SSL_
			if (encryption == SSLENC) {
				LogPrint(0, LOG_NETWORK, 2, "Accepting incoming TCP SSL connection...");
				sslCon = new SSLConnection();
				if (!sslCon->init(sslCertPath.c_str(), sslKeyPath.c_str())) {
					shutdown(new_sock, SD_BOTH);
					closesocket(new_sock);
					delete(sslCon);
					sslCon = NULL;
					LogPrint(0, LOG_NETWORK, 2, "Failed initing incoming TCP SSL connection...");
					continue;
				}
				LogPrint(0, LOG_NETWORK, 2, "SSL connection initialised");
				if (!sslCon->connect(new_sock, localAddress, dataReceiver)) {
					delete(sslCon);
					sslCon = NULL;
					LogPrint(0, LOG_NETWORK, 2, "Failed connecting incoming TCP SSL connection...");
					continue;
				}
				LogPrint(0, LOG_NETWORK, 5, "Successfully accepted incoming TCP SSL connection");
				return sslCon;
			}
			else if (encryption == AESENC) {
			}
			else {
			#endif // _USE_SSL_
				lastActivity = GetTimeNow();
				newCon = new TCPConnection();
				// LogPrint(0, LOG_NETWORK, 1, "Incoming connection: %d... ", (int)new_sock);
				if (!newCon->connect(new_sock, localAddress, dataReceiver)) {
					//printf("FAILED\n");
					LogPrint(0, LOG_NETWORK, 2, "Failed connecting incoming TCP connection...");
					delete(newCon);
				}
				else {
					//printf("SUCCESS\n");
					LogPrint(0, LOG_NETWORK, 5, "Successfully accepted incoming TCP connection");
					return newCon;
				}
			#ifdef _USE_SSL_
			}
			#endif // _USE_SSL_
		}
	}
	return NULL;
}


bool TCPListener::run() {
	NetworkConnection* con;

	if (receiver == NULL)
		return false;

	isRunning = true;
	while (shouldContinue) {
		if ( (con = acceptConnection(100)) != NULL) {
			if (receiver != NULL)
				receiver->receiveNetworkConnection(con);
			else
				delete(con);
		}
	}
	isRunning = false;
	return true;
}








NetworkConnection::NetworkConnection() {
	CHECKNETWORKINIT

	type = 0;
	localAddress = 0;
	remoteAddress = 0;
	threadID = 0;
	lastActivity = 0;
	socket = INVALID_SOCKET;
	remote = false;
	connectTimeoutMS = 0;
	bufferLen = 0;
	buffer = NULL;
	bufferContentLen = 0;
	bufferContentPos = 0;
	receiver = NULL;
	resizeBuffer(INITIALBUFFERSIZE);
	inputSpeed = outputSpeed = 0;
	inputBytes = outputBytes = 0;
	greetingData = NULL;
	greetingSize = 0;
}

NetworkConnection::~NetworkConnection() {
	//printf("~NetworkConnection(%p)\n", this); fflush(stdout);
	disconnect();
	localAddress = 0;
	remoteAddress = 0;
	threadID = 0;
	lastActivity = 0;
	socket = INVALID_SOCKET;
	remote = false;
	if (buffer != NULL)
		delete(buffer);
	bufferLen = 0;
	bufferContentLen = 0;
	bufferContentPos = 0;
	receiver = NULL;
	buffer = NULL;
	if (greetingData)
		delete greetingData;
	greetingData = NULL;
	greetingSize = 0;
}

bool NetworkConnection::setGreetingData(const char* data, uint32 size) {
	if (greetingData)
		delete greetingData;
	if (!data)
		greetingData = NULL;
	else {
		greetingData = new char[size];
		memcpy(greetingData, data, size);
	}
	greetingSize = size;
	return true;
}


bool NetworkConnection::setConnectTimeout(uint32 timeoutMS) {
	connectTimeoutMS = timeoutMS;
	return true;
}

uint64 NetworkConnection::getRemoteAddress() {
	return remoteAddress;
}

bool NetworkConnection::isRemote() {
	return remote;
}

void NetworkConnection::disconnectInternal(uint16 error) {
	LogPrint(0, LOG_NETWORK, 5, "Shutting down network connection with code: %u", error);
	stop();
	if (socket != INVALID_SOCKET) {
		shutdown(socket, SD_BOTH);
		try {
			closesocket(socket);
		}
		catch (...) {}
		socket = INVALID_SOCKET;
	}
	bufferContentLen = 0;
	bufferContentPos = 0;
	if (receiver && error)
		receiver->registerError(error, this);
}

bool NetworkConnection::disconnect(uint16 error) {
	mutex.enter(1000);
	disconnectInternal(error);
	mutex.leave();
	return true;
}

bool NetworkConnection::resizeBuffer(uint32 len) {
	if (len < 128) return false;
	char* newBuffer = new char[len];
	if (bufferContentLen != bufferContentPos) {
		memcpy(newBuffer, buffer+bufferContentPos, bufferContentLen-bufferContentPos);
		bufferContentLen -= bufferContentPos;
		bufferContentPos = 0;
	}
	else {
		bufferContentLen = 0;
		bufferContentPos = 0;
	}

	if (buffer != NULL)
		delete [] buffer;
	bufferLen = len;
	buffer = newBuffer;
	return true;
}

//int32 NetworkConnection::peekStream() {
//
//	mutex.enter();
//
//	// and read from the socket
//	int count = ::recvfrom(socket,buffer+bufferContentLen,bufferLen-bufferContentLen,MSG_PEEK,NULL,0);
//	if(count==SOCKET_ERROR) {
//		int err = utils::GetLastOSErrorNumber();
//		mutex.leave();
//		if ( (err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN) )
//			return 0;
//		else {
//			disconnect(NETWORKERROR_RECEIVE);
//			return -1;
//		}
//	}
//	mutex.leave();
//	return count;
//}

int32 NetworkConnection::readIntoBuffer() {

	// Assume that the mutex is locked
	// LogPrint(0,0,0,"<<<<<<< READINTOBUFFER <<<<<<<<\n");

#ifdef TCPCON_PRINT_DEBUG
	uint64 start = GetTimeNow();
#endif
	int c = 0;
	int32 count = 0;
	// The socket is put in non-blocking mode once at connect/accept and is never
	// flipped back (see TCPConnection::send), so we no longer re-set it here -
	// that was two fcntl() syscalls on every read of the receive hot path.
	// Read from the socket while data is still available
	do {
		//LogPrint(0,0,0,"<<<<<<< READINTOBUFFER RECVFROM start <<<<<<<<\n");
		c = ::recvfrom(socket,buffer+bufferContentLen,bufferLen-bufferContentLen,0,NULL,0);
	//	c = ::recv(socket,buffer+bufferContentLen,bufferLen-bufferContentLen,0);
	//	LogPrint(0,0,0,"<<<<<<< READINTOBUFFER RECVFROM end %d <<<<<<<<\n", c);
		if(c==SOCKET_ERROR) {
			int err = utils::GetLastOSErrorNumber();
			if ( (err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN) )
				break;
			else {
				#ifdef WINDOWS
					if (err != WSAENOTSOCK)
				#endif
					disconnectInternal(NETWORKERROR_RECEIVE);
				return -1;
			}
		}
		if (!c) {
			bool dataAvailable = utils::WaitForSocketReadability(socket, (uint32)(10));
			if (dataAvailable) {
				// this indicates that the socket is in a CLOSE_WAIT state,
				// i.e. the other end has closed the socket so we should too...
				LogPrint(0, LOG_NETWORK, 2, "*** SOCKET DISCONNECT DETECTED ***\n");
				disconnectInternal(NETWORKERROR_RECEIVE);
			}
		}

		if (c) {
			//rounds++;
			count += c;
			bufferContentLen += c;

			if (bufferLen-bufferContentLen < 512) {
				if (bufferLen-bufferContentLen + bufferContentPos > 512) {
					//printf("x"); fflush(stdout);
					memmove(buffer, buffer+bufferContentPos, bufferContentLen-bufferContentPos);
					bufferContentLen -= bufferContentPos;
					bufferContentPos = 0;
				}
				else {
					//printf("o"); fflush(stdout);
					resizeBuffer(bufferLen * 5);
				}
				//resize++;
			}
		}
	} while (c > 0);

	#ifdef TCPCON_PRINT_DEBUG
		if (this->type == TCPCON) {
			if (count) {
				char* tmp = new char[count+1];
				memcpy(tmp, buffer+bufferContentLen-count, count);
				tmp[count] = 0;
				LogPrint(0,LOG_NETWORK,0,"<<<<<<< READINTOBUFFER <<<<<<<< TCP RECV BUF %d bytes '%s' (%.3fms) %d / %d\n", count, tmp, GetTimeAge(start)/1000.0, rounds, resize);
				delete [] tmp;
			}
			else {
				// LogPrint(0,LOG_NETWORK,0,"<<<<<<< READINTOBUFFER <<<<<<<< NOTHING\n");
			}
		}
	#endif

	#ifdef UDPCON_PRINT_DEBUG
		if (this->type == UDPCON) {
			if (count) {
				char* tmp = new char[count+1];
				memcpy(tmp, buffer+bufferContentLen-count, count);
				tmp[count] = 0;
				LogPrint(0,LOG_NETWORK,0,"<<<<<<< READINTOBUFFER <<<<<<<< UDP RECV BUF %u bytes '%s'\n", count, tmp);
				delete [] tmp;
			}
			else {
			//	LogPrint(0,LOG_NETWORK,0,"<<<<<<< READINTOBUFFER <<<<<<<< UDP NOTHING\n");
			}
		}
	#endif

	return count;
}

bool NetworkConnection::receiveAvailable(char* data, uint32& size, uint32 maxSize, uint32 timeout, bool peek) {

	mutex.enter(1000);
	if (maxSize > bufferLen-bufferContentLen) {
		if (bufferContentLen + bufferContentPos > maxSize) {
			memmove(buffer, buffer+bufferContentPos, bufferContentLen-bufferContentPos);
			bufferContentLen -= bufferContentPos;
			bufferContentPos = 0;
		}
		else
			resizeBuffer(bufferLen + maxSize + INITIALBUFFERSIZE);
	}

	uint64 start = GetTimeNow();
	uint32 count = 0;
	bool dataAvailable;

	int32 c = readIntoBuffer();
	if (c < 0) {
		mutex.leave();
		return false;
	}

	if (timeout > 0) {

		// Wait for full size to be available, if not already
		int32 timespent = 0;
		if (bufferContentLen - bufferContentPos < maxSize) {
			while (true) {
				mutex.leave();
				dataAvailable = utils::WaitForSocketReadability(socket, (uint32)(timeout-timespent));
				mutex.enter(1000);
				if ((c = readIntoBuffer()) < 0) {
					mutex.leave();
					return false;
				}
				if (!c && dataAvailable) {
					// this indicates that the socket is in a CLOSE_WAIT state,
					// i.e. the other end has closed the socket so we should too...
					LogPrint(0, LOG_NETWORK, 2, "*** SOCKET DISCONNECT DETECTED ***\n");
					disconnectInternal(NETWORKERROR_RECEIVE);
				}
				if (bufferContentLen - bufferContentPos >= maxSize)
					break;
				if ((timespent = GetTimeAgeMS(start)) >= (int32)timeout)
					break;
			}
		}
	}

	size = bufferContentLen - bufferContentPos;
	if (size > maxSize)
		size = maxSize;

	inputBytes += size;
	if ( size > 0 ) {
		memcpy(data, buffer+bufferContentPos, size);
		if (!peek) {
			bufferContentPos += size;
			if (bufferContentPos == bufferContentLen)
				bufferContentPos = bufferContentLen = 0;
		}
	}
	mutex.leave();
	return true;
}

bool NetworkConnection::receive(char* data, uint32 size, uint32 timeout, bool peek) {
	// Exact-size, timeout-bounded read from the shared receive buffer. The
	// buffer mutex makes this safe against the connection's reader thread;
	// while more bytes are needed the mutex is RELEASED around the
	// WaitForSocketReadability() select() so writers/other readers are not
	// blocked for the whole timeout. With peek=true the bytes are copied but
	// left in the buffer — protocol handlers use this to sniff frame headers
	// (e.g. MessageProtocol peeks size+id before allocating the body buffer).

	mutex.enter(1000);

	uint64 start = GetTimeNow();

	// Only read from the socket if we don't already have enough buffered. A single
	// readIntoBuffer() drains everything currently available - often the whole
	// message, or several pipelined messages, in one go - so the body read (and
	// subsequent messages) are then served straight from the buffer with no
	// recvfrom() syscall at all. Previously readIntoBuffer() ran on every receive()
	// regardless, costing a wasted syscall per peek and per body read.
	if (bufferContentLen - bufferContentPos < size) {
		int32 c = readIntoBuffer();
		if (c < 0) {
			mutex.leave();
			return false;
		}

		// Wait for the full size to be available, if not already.
		int32 timespent = 0;
		while (bufferContentLen - bufferContentPos < size) {
			mutex.leave();
			utils::WaitForSocketReadability(socket, (uint32)(timeout-timespent));
			mutex.enter(1000);
			if ((c = readIntoBuffer()) < 0) {
				mutex.leave();
				return false;
			}
			if (bufferContentLen - bufferContentPos >= size)
				break;
			if ((timespent = GetTimeAgeMS(start)) >= (int32)timeout) {
				mutex.leave();
				return false;
			}
		}
	}

	// LogPrint(0,LOG_NETWORK,0,"Receiving %u (buffer %u) bytes via network...", size, bufferContentLen - bufferContentPos);
	//printf("**************** Buffer ok (%u, %u = %u->%u), left (%u), size (%u), next %u, %u ****************\n",
	//	bufferLen, bufferContentLen - bufferContentPos, bufferContentPos, bufferContentLen,
	//	bufferLen-bufferContentLen+bufferContentPos, size,
	//	*(uint32*)(buffer+bufferContentPos), *(uint32*)(buffer+bufferContentPos+4));

	int64 t = GetTimeAge(start);
	if (t > 0)
		inputSpeed = (uint32)(size*1000000.0/(double)t);
	inputBytes += size;

	memcpy(data, buffer+bufferContentPos, size);
	if (!peek) {
		bufferContentPos += size;
		if (bufferContentPos == bufferContentLen)
			bufferContentPos = bufferContentLen = 0;
	}
	mutex.leave();
	return true;
}

bool NetworkConnection::discard(uint32 size) {
	mutex.enter(1000);
	readIntoBuffer();
	if (bufferContentLen - bufferContentPos >= size) {
		bufferContentPos += size;
		if (bufferContentPos == bufferContentLen)
			bufferContentPos = bufferContentLen = 0;
		inputBytes += size;
		mutex.leave();
		return true;
	}
	else {
		mutex.leave();
		return false;
	}
}

uint32 NetworkConnection::clearBuffer() {
	mutex.enter(1000);
	readIntoBuffer();
	uint32 c = bufferContentLen - bufferContentPos;
	if (c) {
		LogPrint(0,LOG_NETWORK,3,"Cleared %u chars from buffer...", c);
	//	if (c > 1)
	//		utils::PrintBinary(buffer+bufferContentPos, c, false, "Cleared Buffer Content");
	}
	bufferContentPos = bufferContentLen = 0;
	inputBytes += c;
	mutex.leave();
	return c;
}


bool NetworkConnection::waitForDataToRead(uint32 timeout) {
	if (bufferContentLen != bufferContentPos)
		return true;
	else {
		return utils::WaitForSocketReadability(socket, timeout);
	}
}

bool NetworkConnection::waitForDataToBeWritten(uint32 timeout) {
	return utils::WaitForSocketWriteability(socket, timeout);
}


bool NetworkConnection::isConnected(int timeout) {

	if (socket == INVALID_SOCKET)
		return false;

	char peekBuffer[1];
	// First use the socket a bit
	if (!mutex.enter(1000))
		return false;

	utils::SetSocketNonBlockingMode(socket);
	int res = recv(socket, peekBuffer, 1, MSG_PEEK);
	if (res > 0) {
		mutex.leave();
		return true;
	}
	else if (res < 0) {
		int err = utils::GetLastOSErrorNumber();
		// UDP needs TRYAGAIN
		if (err == SOCKETWOULDBLOCK) {
			mutex.leave();
			return true;
		}

		#ifdef WINDOWS
			else if ((type == UDPCON) && (err == SOCKETTRYAGAIN)) {
				mutex.leave();
				return true;
			}
			else if (err == WSAENOTCONN) {} // Windows needs this one...
			else if (err == WSAENOTSOCK) {} // Windows needs this one...
			else if (err == WSAEOPNOTSUPP) {} // WinCE may need this one...???.
		#else
			else if (err == SOCKETTRYAGAIN) {
				mutex.leave();
				return true;
			}
		#endif //WINDOWS
		else {
			// Error will be handled by the caller
			disconnectInternal(0);
			mutex.leave();
			return false;
		}
	//	mutex.leave();
	}
	else { // if err == 0
		// Error will be handled by the caller
		disconnectInternal(0);
		mutex.leave();
		return false;
	}

	struct timeval tv;
	tv.tv_sec = 0;
	tv.tv_usec = 10;

	int maxfd = 0;
	fd_set wfds;
	// create a list of sockets to check for activity
	FD_ZERO(&wfds);
	// specify socket
	FD_SET(socket, &wfds);

	mutex.leave();

	#ifdef WINDOWS
		int len;
	#else
		#ifdef Darwin
			#if GCC_VERSION < 40000
				int len;
				maxfd = socket + 1;
			#else
				socklen_t len;
				maxfd = socket + 1;
			#endif // GCC_VERSION < 40000
		#else
			socklen_t len;
			maxfd = socket + 1;
		#endif
	#endif

	if (timeout > 0) {
		ldiv_t d = ldiv(timeout*1000, 1000000);
		tv.tv_sec = d.quot;
		tv.tv_usec = d.rem;
	}

	// Check for writability
	res = select(maxfd, NULL, &wfds, NULL, &tv);

	if (res <= 0)
		return false;

	//printf("wait res > 0\n");

	int error;
	len = sizeof(error);

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

	if (FD_ISSET(socket, &wfds) != 0) {
		if (getsockopt(socket, SOL_SOCKET, SO_ERROR, (char*)&error, &len) != 0) {

			int wsaError = utils::GetLastOSErrorNumber();
			mutex.leave();
			if (wsaError == 0) {
				// No error, just unable to send...
				return true;
			}
			#ifdef WINDOWS
				else if (wsaError == WSAENETDOWN ) {
					return false;
				}
				else if (wsaError == WSAEFAULT  ) {
					return false;
				}
				else if (wsaError == WSAEINPROGRESS  ) {
					return false;
				}
				else if (wsaError == WSAEINVAL  ) {
					return false;
				}
				else if (wsaError == WSAENOPROTOOPT  ) {
					if (error == 0) {
						return true;
					}
				}
				else if (wsaError == WSAENOTSOCK  ) {
					return false;
				}
			#endif			
			return false;
		}
		mutex.leave();
		if (error == 0)
			return true;
	}
	else
		mutex.leave();

	return false;
}

bool NetworkConnection::didConnect(SOCKET s, int timeout) {
	struct timeval tv;
	tv.tv_sec = 0;
	tv.tv_usec = 10;

	int maxfd = 0;
	fd_set wfds;
	// create a list of sockets to check for activity
	FD_ZERO(&wfds);
	// specify socket
	FD_SET(s, &wfds);


	#ifdef WINDOWS
		int len;
	#else
	#ifdef Darwin
	#if GCC_VERSION < 40000
		int len;
		maxfd = s + 1;
	#else
		socklen_t len;
		maxfd = s + 1;
	#endif // GCC_VERSION < 40000
	#else
		socklen_t len;
		maxfd = s + 1;
	#endif
	#endif

	if (timeout > 0) {
		ldiv_t d = ldiv(timeout * 1000, 1000000);
		tv.tv_sec = d.quot;
		tv.tv_usec = d.rem;
	}

	// Check for writability
	int res = select(maxfd, NULL, &wfds, NULL, &tv);

	if (res <= 0)
		return false;

	//printf("wait res > 0\n");

	int error;
	len = sizeof(error);

	if (FD_ISSET(s, &wfds) != 0) {
		if (getsockopt(s, SOL_SOCKET, SO_ERROR, (char*)&error, &len) != 0) {

			int wsaError = utils::GetLastOSErrorNumber();
			if (wsaError == 0) {
				// No error, just unable to send...
				return true;
			}
		#ifdef WINDOWS
			else if (wsaError == WSAENETDOWN) {
				return false;
			}
			else if (wsaError == WSAEFAULT) {
				return false;
			}
			else if (wsaError == WSAEINPROGRESS) {
				return false;
			}
			else if (wsaError == WSAEINVAL) {
				return false;
			}
			else if (wsaError == WSAENOPROTOOPT) {
				if (error == 0) {
					return true;
				}
			}
			else if (wsaError == WSAENOTSOCK) {
				return false;
			}
		#endif			
			return false;
		}
		if (error == 0)
			return true;
	}
	return false;
}

bool NetworkConnection::didConnect(int timeout) {

	if (socket == INVALID_SOCKET)
		return false;

	// First use the socket a bit
	if (!mutex.enter(1000))
		return false;

	struct timeval tv;
	tv.tv_sec = 0;
	tv.tv_usec = 10;

	int maxfd = 0;
	fd_set wfds;
	// create a list of sockets to check for activity
	FD_ZERO(&wfds);
	// specify socket
	FD_SET(socket, &wfds);

	mutex.leave();

	#ifdef WINDOWS
		int len;
	#else
	#ifdef Darwin
	#if GCC_VERSION < 40000
		int len;
		maxfd = socket + 1;
	#else
		socklen_t len;
		maxfd = socket + 1;
	#endif // GCC_VERSION < 40000
	#else
		socklen_t len;
		maxfd = socket + 1;
	#endif
	#endif

	if (timeout > 0) {
		ldiv_t d = ldiv(timeout * 1000, 1000000);
		tv.tv_sec = d.quot;
		tv.tv_usec = d.rem;
	}

	// Check for writability
	int res = select(maxfd, NULL, &wfds, NULL, &tv);

	if (res <= 0)
		return false;

	//printf("wait res > 0\n");

	int error;
	len = sizeof(error);

	mutex.enter(1000);
	if (FD_ISSET(socket, &wfds) != 0) {
		if (getsockopt(socket, SOL_SOCKET, SO_ERROR, (char*)&error, &len) != 0) {

			int wsaError = utils::GetLastOSErrorNumber();
			mutex.leave();
			if (wsaError == 0) {
				// No error, just unable to send...
				return true;
			}
#ifdef WINDOWS
			else if (wsaError == WSAENETDOWN) {
				return false;
			}
			else if (wsaError == WSAEFAULT) {
				return false;
			}
			else if (wsaError == WSAEINPROGRESS) {
				return false;
			}
			else if (wsaError == WSAEINVAL) {
				return false;
			}
			else if (wsaError == WSAENOPROTOOPT) {
				if (error == 0) {
					return true;
				}
			}
			else if (wsaError == WSAENOTSOCK) {
				return false;
			}
#endif			
			return false;
		}
		mutex.leave();
		if (error == 0)
			return true;
	}
	else
		mutex.leave();

	return false;
}

bool NetworkConnection::run() {
	if (receiver == NULL)
		return false;

	uint32 size;
	uint32 buflen = INITIALBUFFERSIZE;
	char* myBuffer = (char*) malloc(buflen);
	if (myBuffer == NULL)
		return false;

	LogPrint(0, LOG_NETWORK, 2, "Incoming network connection from %u.%u.%u.%u:%u, started run...",
		GETIPADDRESSQUAD(remoteAddress), GETIPPORT(remoteAddress));

	isRunning = true;
	while (shouldContinue) {
		if (receive((char*)&size, sizeof(size), 50, true)) {

			if (buflen < size) {
				buflen = size;
				myBuffer = (char*) realloc(myBuffer, buflen);
				if (myBuffer == NULL) {
					isRunning = false;
					disconnect(NETWORKERROR_MEMORYFULL);
					return false;
				}
			}

			// Receive the full data structure
			if (!receive(myBuffer, size, 500)) {
				isRunning = false;
				free(myBuffer);
				// Error already reported & already disconnected
				return false;
			}

			if (receiver) {
				receiver->receiveData(myBuffer, size, this);
			}
		}
	}
	isRunning = false;
	free(myBuffer);
	return true;
}

uint32 NetworkConnection::getOutputSpeed() {
	return outputSpeed;
}

uint32 NetworkConnection::getInputSpeed() {
	return inputSpeed;
}

uint8 NetworkConnection::getConnectionType() {
	return type;
}











UDPConnection::UDPConnection() : NetworkConnection() {
	type = UDPCON;
	defaultReceiver = 0L;
}

UDPConnection::~UDPConnection() {
}

bool UDPConnection::initForOutputOnly() {
	// Setup socket
	mutex.enter(1000);
	if((socket=::socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP))==INVALID_SOCKET){
		LogPrint(0, LOG_NETWORK, 0, "Could not create UDPConnection socket...");
		disconnectInternal(0);
		mutex.leave();
		return false;
	}
  
	#ifdef WINDOWS
		// Set the exclusive address option, preventing other software binding to
		// non-INADDR_ANY (i.e. interface addresses such as localhost directly)
		int one = 1;
		setsockopt(socket, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, (char *) &one, sizeof(one));
	#else
		/*
			This socket option tells the kernel that even if this port is busy (in
			the TIME_WAIT state), go ahead and reuse it anyway.  If it is busy,
			but with another state, you will still get an address already in use
			error.  It is useful if your server has been shut down, and then
			restarted right away while sockets are still active on its port.  You
			should be aware that if any unexpected data comes in, it may confuse
			your server, but while this is possible, it is not likely.
		*/
		int one = 1;
		setsockopt(socket,SOL_SOCKET,SO_REUSEADDR,&one,sizeof(one));
	#endif

	setsockopt(socket,SOL_SOCKET,SO_BROADCAST,(char*)&one,sizeof(one));

	// Set blocking mode
	utils::SetSocketNonBlockingMode(socket);

	mutex.leave();
	return true;
}

bool UDPConnection::connect(uint16 port, NetworkDataReceiver* receiver) {
	if (port == 0) {
		LogPrint(0, LOG_NETWORK, 0, "Could not start UDPConnection on port 0...");
		return false;
	}

	memcpy(((char*)&localAddress)+sizeof(uint32), &port, sizeof(uint16));
	// Open UDP port

	// Setup listening on port node->networkPort
	mutex.enter(1000);
	if((socket=::socket(AF_INET,SOCK_DGRAM,IPPROTO_UDP))==INVALID_SOCKET){
		LogPrint(0, LOG_NETWORK, 0, "Could not create UDPConnection socket...");
		disconnectInternal(0);
		mutex.leave();
		return false;
	}
  
	#ifdef WINDOWS
		// Set the exclusive address option, preventing other software binding to
		// non-INADDR_ANY (i.e. interface addresses such as localhost directly)
		int one = 1;
		setsockopt(socket, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, (char *) &one, sizeof(one));
	#else
		/*
			This socket option tells the kernel that even if this port is busy (in
			the TIME_WAIT state), go ahead and reuse it anyway.  If it is busy,
			but with another state, you will still get an address already in use
			error.  It is useful if your server has been shut down, and then
			restarted right away while sockets are still active on its port.  You
			should be aware that if any unexpected data comes in, it may confuse
			your server, but while this is possible, it is not likely.
		*/
		int one = 1;
		setsockopt(socket,SOL_SOCKET,SO_REUSEADDR,&one,sizeof(one));
	#endif

	setsockopt(socket,SOL_SOCKET,SO_BROADCAST,(char*)&one,sizeof(one));

	struct	sockaddr_in	addr;
	addr.sin_family= AF_INET;
	addr.sin_addr.s_addr=INADDR_ANY;
	addr.sin_port=htons(port);

	if(bind(socket,(SOCKADDR*)&addr,sizeof(struct sockaddr_in))==SOCKET_ERROR){
		LogPrint(0, LOG_NETWORK, 0, "Could not create UDPConnection on port %u...", port);
		disconnectInternal(0);
		mutex.leave();
		return false;
	}
	LogPrint(0, LOG_NETWORK, 2, "Created UDPConnection on port %u...", port);

	// Set blocking mode
	utils::SetSocketNonBlockingMode(socket);

	if (receiver != NULL) {
		this->receiver = receiver;
		// Start networking thread
		if (!ThreadManager::CreateThread(UDPConnectionRun, this, threadID)) {
			LogPrint(0, LOG_NETWORK, 0, "Could not start UDPConnection thread...");
			disconnectInternal(0);
			mutex.leave();
			return false;
		}
	}

	mutex.leave();
	return true;
}

bool UDPConnection::send(const char* data, uint32 size, uint64 receiver) {
	uint64 start;
	sockaddr_in recvAddr;
	recvAddr.sin_family = AF_INET;
	int64 t;

	if (!receiver && !defaultReceiver) {
		// this should broadcast ##############
		return false;
	}
	else {
		uint64 rec = receiver ? receiver : defaultReceiver;
		uint16 p = GETIPPORT(rec);
		recvAddr.sin_port = htons(p);
		memcpy(&recvAddr.sin_addr.s_addr, &rec, 4);
		if (!sendMutex.enter(500))
			return false;
		start = GetTimeNow();

		uint32 pos = 0;
		int32 n;

		while (true) {
			try {
				n = ::sendto(socket, data + pos, size - pos, 0, (SOCKADDR*)&recvAddr, sizeof(sockaddr_in));
			}
			catch (...) {
				// printf("--- UDP error sending reply (%u)...\n", size);
				disconnect(NETWORKERROR_SEND_ERROR);
				sendMutex.leave();
				return false;
			}
			if (n == SOCKET_ERROR) {
				// printf("--- UDP error sending reply (%u)...\n", size);
				disconnect(NETWORKERROR_SEND_ERROR);
				sendMutex.leave();
				return false;
			}
			pos += n;
			// Have we written everything?
			if (pos >= size)
				break;
			// else wait for writeability
			utils::WaitForSocketWriteability(socket, 500);
			// check for ridiculous timeout
			if ((t = GetTimeAgeMS(start)) > 10000) {
				disconnect(NETWORKERROR_SEND_TIMEOUT);
				sendMutex.leave();
				return false;
			}
			// and go again
		}
	}
	t = GetTimeAge(start);
	if (t > 0)
		outputSpeed = (uint32)(size*1000000.0/(double)t);
	outputBytes += size;
	sendMutex.leave();

	#ifdef UDPCON_PRINT_DEBUG
		if (size < 1024) {
			char* tmp = new char[size+1];
			memcpy(tmp, data, size);
			tmp[size] = 0;
			LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> UDP Sent %u bytes '%s' to %u.%u.%u.%u:%u\n", size, tmp, GETIPADDRESSQUAD(receiver), GETIPPORT(receiver));
			delete [] tmp;
		}
		else {
			#ifdef UDPCON_PRINTBINARY_DEBUG
				char* tmp = new char[size+1];
				memcpy(tmp, data, size);
				tmp[size] = 0;
				LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> UDP Sent %u bytes '%s'\n", size, tmp);
				delete [] tmp;
			#else
				char* tmp = new char[1024];
				memcpy(tmp, data, 1023);
				tmp[1023] = 0;
				LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> UDP Sent %u bytes '%s' to %u.%u.%u.%u:%u\n", size, tmp, GETIPADDRESSQUAD(receiver), GETIPPORT(receiver));
				delete [] tmp;
			#endif
		}
	#endif

	return true;
}

bool UDPConnection::reconnect(uint32 timeoutMS) {
	disconnect();
	return connect(GETIPPORT(localAddress), receiver);
}

bool UDPConnection::setDefaultReceiver(uint64 receiver) {
	defaultReceiver = receiver;
	return true;
}

uint64 UDPConnection::getLocalAddress() {
	return localAddress;
}













TCPConnection::TCPConnection() : NetworkConnection() {
	type = TCPCON;
}

TCPConnection::~TCPConnection() {
}

bool TCPConnection::connect(SOCKET s, uint64 localAddr, NetworkDataReceiver* receiver) {
	mutex.enter(1000);
	bufferContentLen = 0;
	bufferContentPos = 0;
	localAddress = localAddr;

	socket = s;
	findRemoteAddress(remoteAddress);

	// Set blocking mode
	utils::SetSocketNonBlockingMode(socket);

	struct linger tmp = {1, 0};
	setsockopt(socket, SOL_SOCKET, SO_LINGER, (char *)&tmp, sizeof(tmp));
	int delay = 1;
	setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, (char*) &delay, sizeof(delay));
	//char buffsize = 1;
	//setsockopt(socket, SOL_SOCKET, SO_SNDBUF, &buffsize, sizeof(buffsize));
	//buffsize = 1;
	//setsockopt(socket, SOL_SOCKET, SO_RCVBUF, &buffsize, sizeof(buffsize));

	LogPrint(0, LOG_NETWORK, 2, "Incoming TCP connection from %u.%u.%u.%u:%u, starting run...",
		GETIPADDRESSQUAD(remoteAddress), GETIPPORT(remoteAddress));

	if (receiver != NULL) {
		this->receiver = receiver;
		// Start networking thread
		if (!ThreadManager::CreateThread(TCPConnectionRun, this, threadID)) {
			LogPrint(0, LOG_NETWORK, 0, "Could not start TCPConnection thread...");
			disconnectInternal(0);
			mutex.leave();
			return false;
		}
	}

	remote = true;
	mutex.leave();
	return true;
}

bool TCPConnection::connect(uint64 addr, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	// Timeout-aware connect, identical on Winsock and BSD sockets thanks to the
	// SOCKET/closesocket/GetLastOSErrorNumber abstractions: the socket is put
	// into non-blocking mode first, so ::connect() returns immediately with
	// EWOULDBLOCK/EINPROGRESS and didConnect() then select()-waits up to
	// timeoutMS for writability (= connection established). On success the
	// socket gets SO_LINGER {1,0} (hard close, RST instead of TIME_WAIT) and
	// TCP_NODELAY (no Nagle batching — vital for small request/reply frames).
	// Passing a NetworkDataReceiver switches the connection to push mode by
	// starting a dedicated reader thread (TCPConnectionRun).

	// first create a temporary socket so we don't have to block the mutex while connecting
	SOCKET tempSocket;

	if (timeoutMS) {
		if (!connectTimeoutMS)
			connectTimeoutMS = timeoutMS;
	}
	else
		timeoutMS = connectTimeoutMS;

	if((tempSocket =::socket(AF_INET,SOCK_STREAM,IPPROTO_TCP))==INVALID_SOCKET){
		int retries = 0;
		int err = utils::GetLastOSErrorNumber();
		while ((err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN)) {
			utils::Sleep(20);
			if ((tempSocket = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP)) != INVALID_SOCKET)
				break;
			err = utils::GetLastOSErrorNumber();
			retries++;
			if (retries > 10)
				break;
		}
		if (tempSocket == INVALID_SOCKET) {
			// mutex is not locked
			LogPrint(0, LOG_NETWORK, 0, "Could not create TCPConnection socket (%d)...", err);
			return false;
		}
	}

	bufferContentLen = 0;
	bufferContentPos = 0;

	// Set blocking mode
	utils::SetSocketNonBlockingMode(tempSocket);

	struct sockaddr_in saServer;
	saServer.sin_family = AF_INET;
	saServer.sin_port = htons(GETIPPORT(addr));
	memcpy(&saServer.sin_addr.s_addr, &addr, 4);

	// Connect to the server
	int res;
	if ((res = ::connect(tempSocket, (struct sockaddr*)&saServer, sizeof(struct sockaddr))) != 0) {
		int err = utils::GetLastOSErrorNumber();
		if ( (err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN) ) {
			if (!didConnect(tempSocket, timeoutMS)) {
				closesocket(tempSocket);
				return false;
			}
		}
		else {
			closesocket(tempSocket);
			return false;
		}
	}

	// Set blocking mode
	// utils::SetSocketNonBlockingMode(socket);

	struct linger tmp = {1, 0};
	setsockopt(tempSocket, SOL_SOCKET, SO_LINGER, (char *)&tmp, sizeof(tmp));
	int delay = 1;
	setsockopt(tempSocket, IPPROTO_TCP, TCP_NODELAY, (char*) &delay, sizeof(delay));
	//char buffsize = 1;
	//setsockopt(tempSocket, SOL_SOCKET, SO_SNDBUF, &buffsize, sizeof(buffsize));
	//buffsize = 1;
	//setsockopt(tempSocket, SOL_SOCKET, SO_RCVBUF, &buffsize, sizeof(buffsize));

	// Now block the mutex
	if (!mutex.enter(1000)) {
		LogPrint(0, LOG_NETWORK, 0, "Could not lock connection mutex...");
		closesocket(tempSocket);
		return false;
	}

	socket = tempSocket;

	if (receiver != NULL) {
		this->receiver = receiver;
		// Start networking thread
		if (!ThreadManager::CreateThread(TCPConnectionRun, this, threadID)) {
			LogPrint(0, LOG_NETWORK, 0, "Could not start TCPConnection thread...");
			disconnectInternal(0);
			mutex.leave();
			return false;
		}
	}

	remoteAddress = addr;
	remote = false;

	mutex.leave();
	return true;
}

bool TCPConnection::connect(const char* addr, uint16 port, uint64& location, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	// Resolve address
	uint32 address;
	if (!utils::LookupIPAddress(addr, address))
		return false;
	return connect(location = GETIPADDRESSPORT(address, port), timeoutMS, receiver);
}

bool TCPConnection::connect(const uint32* addresses, uint16 addressCount, uint16 port, uint64& location, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	// Try all addresses
	for (uint16 n=0; n<addressCount; n++) {
		if (connect(location = GETIPADDRESSPORT(addresses[n], port), timeoutMS, receiver))
			return true;
	}
	return false;
}


bool TCPConnection::delayedConnect(uint64 addr, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	mutex.enter(1000);

	connectTimeoutMS = timeoutMS;
	remoteAddress = addr;
	remote = false;

	if (receiver != NULL) {
		this->receiver = receiver;
		// Start networking thread
		if (!ThreadManager::CreateThread(TCPConnectionRun, this, threadID)) {
			LogPrint(0, LOG_NETWORK, 0, "Could not start TCPConnection thread...");
			mutex.leave();
			return false;
		}
	}

	mutex.leave();
	return true;
}

bool TCPConnection::delayedConnect(const char* addr, uint16 port, uint64& location, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	// Resolve address
	uint32 address;
	if (!utils::LookupIPAddress(addr, address))
		return false;
	return delayedConnect(location = GETIPADDRESSPORT(address, port), timeoutMS, receiver);
}


//#define TCPCON_PRINT_DEBUG

bool TCPConnection::send(const char* data, uint32 size, uint64 receiver) {
	// Ignore receiver, only for UDP
	// ### Consider splitting large data up into smaller chunks ###
	if (!size)
		return false;
	if (!sendMutex.enter(500))
		return false;
	uint64 start = GetTimeNow();
	// The socket stays in non-blocking mode (set once at connect/accept). We used
	// to flip it to blocking for the duration of the send and back to non-blocking
	// afterwards, which cost four fcntl() syscalls on every message AND raced with
	// the receive thread reading the same socket. Instead we drive the non-blocking
	// send loop directly and only wait for writeability when the kernel send buffer
	// is genuinely full (EWOULDBLOCK).

	// Charles:
			//const char *data = (const char *) tdata;
			//	IntT done = 0;
			//	do {
			//	  int n = write(fd,&(data[done]),length - done);
			//	  if(n < 0) {
			// if(errno == EAGAIN || errno == EINTR) // Recoverable error?
			//   continue;
			// return -1;
			//	  }
			//	  done += n;
			//	} while(done < length);
			//	return done;

	int64 t;
	uint32 pos = 0;
	int32 n;

	while (true) {
		try {
			n = ::send(socket, data + pos, size - pos, 0);
		}
		catch (...) {
			//printf("--- TCP error sending reply (%u) error: %u...\n", size, WSAGetLastError());
			disconnect(NETWORKERROR_SEND_ERROR);
			sendMutex.leave();
			return false;
		}
		//n = ::send(socket, data + pos,
		//	(size - pos>5120000) ? 5120000 : size - pos,
		//	0);
		if(n == SOCKET_ERROR) {
			int err = utils::GetLastOSErrorNumber();
			if ((err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN)) {
				// Kernel send buffer is full - wait until it drains, then retry
				// the same chunk. Nothing has been consumed, so 'pos' is unchanged.
				if ((t = GetTimeAgeMS(start)) > 10000) {
					disconnect(NETWORKERROR_SEND_TIMEOUT);
					sendMutex.leave();
					return false;
				}
				utils::WaitForSocketWriteability(socket, 1000);
				continue;
			}
			//printf("--- TCP error sending reply (%u) error: %u...\n", size, WSAGetLastError());
			disconnect(NETWORKERROR_SEND_ERROR);
			sendMutex.leave();
			return false;
		}
		pos += n;
		// Have we written everything?
		if (pos >= size)
			break;
		// check for ridiculous timeout
		if ((t = GetTimeAgeMS(start)) > 10000) {
			disconnect(NETWORKERROR_SEND_TIMEOUT);
			sendMutex.leave();
			return false;
		}
		// and go again
	}

	t = GetTimeAge(start);
	if (t > 0)
		outputSpeed = (uint32)(size*1000000.0/(double)t);
	outputBytes += size;
	sendMutex.leave();

	#ifdef TCPCON_PRINT_DEBUG
		if (size < 1024) {
			char* tmp = new char[size+1];
			memcpy(tmp, data, size);
			tmp[size] = 0;
			LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> TCP Sent %u bytes (%.3f) '%s'\n", size, t/1000.0, tmp);
			delete [] tmp;
		}
		else {
			#ifdef TCPCON_PRINTBINARY_DEBUG
				char* tmp = new char[size+1];
				memcpy(tmp, data, size);
				tmp[size] = 0;
				LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> TCP Sent %u bytes '%s'\n", size, tmp);
				delete [] tmp;
			#else
				char* tmp = new char[1024];
				memcpy(tmp, data, 1023);
				tmp[1023] = 0;
				LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> TCP Sent %u bytes (%.3f) '%s'\n", size, t / 1000.0, tmp);
				delete [] tmp;
			#endif
		}
	#endif


//	utils::Sleep(100);
	return true;
}



bool TCPConnection::reconnect(uint32 timeoutMS) {
	disconnect();
	return connect(remoteAddress, timeoutMS, receiver);
}


bool TCPConnection::findRemoteAddress(uint64& addr) {

	if (socket == INVALID_SOCKET)
		return false;

	struct sockaddr_in remoteAddr;
	
	#ifdef WINDOWS
		int remoteAddrLen;
	#else
		#ifdef Darwin
			#if GCC_VERSION < 40000
				int remoteAddrLen;
			#else
				socklen_t remoteAddrLen;
			#endif // GCC_VERSION < 40000
		#else
			socklen_t remoteAddrLen;
		#endif
	#endif // WINDOWS

	remoteAddrLen = sizeof(struct sockaddr_in);

	if (getpeername(socket, (struct sockaddr*) &remoteAddr, &remoteAddrLen) != 0)
		return false;

	uint32 address = remoteAddr.sin_addr.s_addr;
	if ( address == LOCALHOSTIP ) {
		// Get the actual local ip address, if possible
		utils::GetLocalIPAddress(address);
	}
	addr = GETIPADDRESSPORT(address, (uint16)remoteAddr.sin_port);
	return true;
}
















bool IsSSLInitialised = false;

#ifdef _USE_SSL_
#ifdef WINDOWS
	// #include "openssl/applink.c"
#endif
#endif


SSLConnection::SSLConnection() : NetworkConnection() {
	type = SSLCON;
	allowSelfSigned = DefaultAllowSelfSigned;
	caFile = DefaultCAFile;
	caPath = DefaultCAPath;
	#ifdef _USE_SSL_
		if (!IsSSLInitialised) {
			// Init SSL
			//CRYPTO_malloc_init();
			OpenSSL_add_all_algorithms();
			//ERR_load_BIO_strings();
			ERR_load_crypto_strings();
			SSL_load_error_strings();

			certbio = BIO_new(BIO_s_file());
			outbio  = BIO_new_fp(stdout, BIO_NOCLOSE);

			SSL_library_init();

			IsSSLInitialised = true;
		}
		ctx = NULL;
		ssl = NULL;
	#endif // _USE_SSL_
}

SSLConnection::~SSLConnection() {
	#ifdef _USE_SSL_
		mutex.enter(1000);
		if (ctx) {
			SSL_CTX_free(ctx);
			ctx = NULL;
		}
		mutex.leave();
	#else // _USE_SSL_
	#endif // _USE_SSL_
}

// Process-wide default for client certificate verification. False = verify
// peers against the CA trust store (secure by default). Set from the global
// <psyspec allowselfsigned="yes"> attribute; per-connection
// setAllowSelfSigned() overrides it either way.
bool SSLConnection::DefaultAllowSelfSigned = false;

void SSLConnection::SetDefaultAllowSelfSigned(bool allow) {
	DefaultAllowSelfSigned = allow;
}

bool SSLConnection::GetDefaultAllowSelfSigned() {
	return DefaultAllowSelfSigned;
}

// Process-wide default custom CA location (set from <psyspec cafile/capath>);
// per-connection setCALocation() overrides it. Empty = use OS trust store.
std::string SSLConnection::DefaultCAFile;
std::string SSLConnection::DefaultCAPath;

void SSLConnection::SetDefaultCALocation(const char* caFile, const char* caPath) {
	DefaultCAFile = caFile ? caFile : "";
	DefaultCAPath = caPath ? caPath : "";
}

void SSLConnection::setCALocation(const char* caFile, const char* caPath) {
	this->caFile = caFile ? caFile : "";
	this->caPath = caPath ? caPath : "";
	#ifdef _USE_SSL_
		if (ctx)
			applyClientVerify();
	#endif // _USE_SSL_
}

void SSLConnection::setVerifyHostName(const char* host) {
	verifyHostName = host ? host : "";
}

void SSLConnection::setAllowSelfSigned(bool allow) {
	allowSelfSigned = allow;
	#ifdef _USE_SSL_
		if (ctx)
			applyClientVerify();
	#endif // _USE_SSL_
}

#ifdef _USE_SSL_
// Configure peer verification on the client context according to the
// allowSelfSigned policy for this connection.
bool SSLConnection::applyClientVerify() {
	if (!ctx)
		return false;
	if (allowSelfSigned) {
		SSL_CTX_set_verify(ctx, SSL_VERIFY_NONE, NULL);
	}
	else {
		if (!caFile.empty() || !caPath.empty()) {
			// Verify the server certificate against a custom CA file/dir
			// (e.g. a self-minted CA) instead of the OS trust store
			if (SSL_CTX_load_verify_locations(ctx,
					caFile.empty() ? NULL : caFile.c_str(),
					caPath.empty() ? NULL : caPath.c_str()) != 1)
				LogPrint(0, LOG_NETWORK, 0, "SSL: could not load CA location (file: %s, path: %s)",
					caFile.empty() ? "-" : caFile.c_str(), caPath.empty() ? "-" : caPath.c_str());
		}
		else {
			// Verify the server certificate against the OS/CA trust store
			if (SSL_CTX_set_default_verify_paths(ctx) != 1)
				LogPrint(0, LOG_NETWORK, 0, "SSL: could not load default CA trust store");
		}
		SSL_CTX_set_verify(ctx, SSL_VERIFY_PEER, NULL);
	}
	return true;
}

int SSLConnection::getVerifyMode() {
	if (!ctx)
		return -1;
	return SSL_CTX_get_verify_mode(ctx);
}
#endif // _USE_SSL_

bool SSLConnection::init(const char *certFile, const char *keyFile) {
	mutex.enter(1000);
	#ifdef _USE_SSL_
		LogPrint(0, LOG_NETWORK, 2, "SSL connection init start");
		// Compatible with SSLv2, SSLv3 and TLSv1
		const SSL_METHOD *method = SSLv23_server_method();
		// Create new context from method.
		ctx = SSL_CTX_new(method);
		if(!ctx) {
			LogPrint(0, LOG_NETWORK, 0, "Unable to create a new SSL context structure");
			BIO_printf(outbio, "Unable to create a new SSL context structure.\n");
			mutex.leave();
			return false;
		}
		LogPrint(0, LOG_NETWORK, 2, "SSL connection init created, setting files...");

		if ( SSL_CTX_use_certificate_chain_file(ctx, certFile) <= 0) {
			LogPrint(0, LOG_NETWORK, 0, "Unable to use SSL certificate chain file: %s", certFile);
			LogSSLErrors("certificate chain file", 0);
			mutex.leave();
			return false;
		}
		if ( SSL_CTX_use_PrivateKey_file(ctx, keyFile, SSL_FILETYPE_PEM) <= 0) {
			LogPrint(0, LOG_NETWORK, 0, "Unable to use SSL private key file: %s", keyFile);
			LogSSLErrors("private key file", 0);
			mutex.leave();
			return false;
		}
		LogPrint(0, LOG_NETWORK, 2, "SSL connection init files set");

		// Verify that the two keys goto together.
		if ( !SSL_CTX_check_private_key(ctx) ) {
			LogPrint(0, LOG_NETWORK, 0, "SSL private key invalid: %s", keyFile);
			fprintf(stderr, "Private key is invalid.\n");
			mutex.leave();
			return false;
		}
		mutex.leave();
		LogPrint(0, LOG_NETWORK, 2, "SSL connection init done");
		return true;
	#else // _USE_SSL_
		mutex.leave();
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::init() {
	mutex.enter(1000);
	#ifdef _USE_SSL_
		// Compatible with SSLv2, SSLv3 and TLSv1
		const SSL_METHOD *method = SSLv23_method();
		// Create new context from method.
		ctx = SSL_CTX_new(method);
		if(!ctx) {
			BIO_printf(outbio, "Unable to create a new SSL context structure.\n");
			mutex.leave();
			return false;
		}

		//SSL_CTX_set_options(ctx, SSL_OP_NO_SSLv2);

		// Client-side certificate verification (secure by default; see
		// setAllowSelfSigned/SetDefaultAllowSelfSigned)
		applyClientVerify();

		mutex.leave();
		return true;
	#else // _USE_SSL_
		mutex.leave();
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::didConnect(int timeout) {
	return NetworkConnection::didConnect(timeout);
}

bool SSLConnection::isConnected(int timeout) {
	
	// ######################
	//return true;
	
	//printf("ISCON1\n");fflush(stdout);

#ifdef _USE_SSL_

	if ((socket == INVALID_SOCKET) || !ssl)
		return false;

	char peekBuffer[1];
	// First use the socket a bit

	//utils::Sleep(100);

	if (!mutex.enter(200, __FUNCTION__))
		return false;
	//printf("ISCON2\n");fflush(stdout);
	int res = recv(socket, peekBuffer, 1, MSG_PEEK);
	if (res > 0) {
		mutex.leave();
		//printf("ISCON3\n");fflush(stdout);
		return true;
	}
	else if (res < 0) {
		int err = utils::GetLastOSErrorNumber();
		// UDP needs TRYAGAIN
		if (err == SOCKETWOULDBLOCK) {
			mutex.leave();
			//printf("ISCON4\n");fflush(stdout);
			return true;
		}

		#ifdef WINDOWS
			else if ((type == UDPCON) && (err == SOCKETTRYAGAIN)) {
				mutex.leave();
				//printf("ISCON5\n");fflush(stdout);
				return true;
			}
			else if (err == WSAENOTCONN) {} // Windows needs this one...
			else if (err == WSAEOPNOTSUPP) {} // WinCE may need this one...???.
		#else
			else if (err == SOCKETTRYAGAIN) {
				mutex.leave();
				return true;
			}
		#endif //WINDOWS
		else {
			// Error will be handled by the caller
			//printf("ISCON6\n");fflush(stdout);
			disconnect(0);
			mutex.leave();
			return false;
		}
	//	mutex.leave();
	}
	else { // if err == 0
		// Error will be handled by the caller
		//printf("ISCON7\n");fflush(stdout);
		disconnect(0);
		mutex.leave();
		return false;
	}

	//printf("ISCON10\n");fflush(stdout);

	struct timeval tv;
	tv.tv_sec = 0;
	tv.tv_usec = 10;

	int maxfd = 0;
	fd_set wfds;
	// create a list of sockets to check for activity
	FD_ZERO(&wfds);
	// specify socket
	FD_SET(socket, &wfds);

	mutex.leave();

	#ifdef WINDOWS
		int len;
	#else
		#ifdef Darwin
			#if GCC_VERSION < 40000
				int len;
				maxfd = socket + 1;
			#else
				socklen_t len;
				maxfd = socket + 1;
			#endif // GCC_VERSION < 40000
		#else
			socklen_t len;
			maxfd = socket + 1;
		#endif
	#endif

	if (timeout > 0) {
		ldiv_t d = ldiv(timeout*1000, 1000000);
		tv.tv_sec = d.quot;
		tv.tv_usec = d.rem;
	}

	// Check for writability
	//printf("ISCON11\n");fflush(stdout);
	res = select(maxfd, NULL, &wfds, NULL, &tv);
	//printf("ISCON12\n");fflush(stdout);

	if (res <= 0)
		return false;

	//printf("wait res > 0\n");

	int error;
	len = sizeof(error);

	mutex.enter(1000);
	//printf("ISCON13\n");fflush(stdout);
	if (FD_ISSET(socket, &wfds) != 0) {
		if (getsockopt(socket, SOL_SOCKET, SO_ERROR, (char*)&error, &len) != 0) {

			int wsaError = utils::GetLastOSErrorNumber();
			mutex.leave();
			if (wsaError == 0) {
				// No error, just unable to send...
				return true;
			}
			#ifdef WINDOWS
				else if (wsaError == WSAENETDOWN ) {
					return false;
				}
				else if (wsaError == WSAEFAULT  ) {
					return false;
				}
				else if (wsaError == WSAEINPROGRESS  ) {
					return false;
				}
				else if (wsaError == WSAEINVAL  ) {
					return false;
				}
				else if (wsaError == WSAENOPROTOOPT  ) {
					if (error == 0) {
						return true;
					}
				}
				else if (wsaError == WSAENOTSOCK  ) {
					return false;
				}
			#endif			
			return false;
		}
		mutex.leave();
		if (error == 0)
			return true;
	}
	else
		mutex.leave();

#endif //_USE_SSL_
	return false;
}

bool SSLConnection::disconnect(uint16 error) {
	#ifdef _USE_SSL_
		mutex.enter(1000);
		if (!ctx || !ssl) {
			mutex.leave();
			return false;
		}
		SSL_shutdown(ssl);
		SSL_free(ssl);
		ssl = NULL;
		disconnectInternal(error);
		mutex.leave();
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::connect(SOCKET s, uint64 localAddr, NetworkDataReceiver* receiver) {
	#ifdef _USE_SSL_
		if (!ctx || !(ssl = SSL_new(ctx)))
			return false;

		mutex.enter(1000);
		bufferContentLen = 0;
		bufferContentPos = 0;
		localAddress = localAddr;

		socket = s;
		findRemoteAddress(remoteAddress);

		// Set blocking mode
		utils::SetSocketNonBlockingMode(socket);

		struct linger tmp = {1, 0};
		setsockopt(s, SOL_SOCKET, SO_LINGER, (char *)&tmp, sizeof(tmp));
		int delay = 1;
		setsockopt(s, IPPROTO_TCP, TCP_NODELAY, (char*) &delay, sizeof(delay));

		LogPrint(0, LOG_NETWORK, 2, "SSL connection options set, accepting connection...");

		remote = true;
		SSL_set_fd(ssl, (int)socket);
		int ret = SSL_accept(ssl);
		int err;
		int errCount = 0;

		while (ret == -1) {
			err = SSL_get_error(ssl, ret);
			if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {

				if ((++errCount) > 10) {
					LogPrint(0, LOG_NETWORK, 1, "SSL connection accept took too long, disconnecting...");
					SSL_shutdown(ssl);
					SSL_free(ssl);
					ssl = NULL;
					disconnectInternal(0);
					mutex.leave();
					return false;
				}

				utils::WaitForSocketReadability(s, 200);
				ret = SSL_accept(ssl);
			}
			else {
				SSL_shutdown(ssl);
				SSL_free(ssl);
				ssl = NULL;
				disconnectInternal(0);
				mutex.leave();
				return false;
			}
		}

		LogPrint(0, LOG_NETWORK, 2, "SSL connection accepted, setting up receiver...");

		if (receiver != NULL) {
			this->receiver = receiver;
			// Start networking thread
			if (!ThreadManager::CreateThread(SSLConnectionRun, this, threadID)) {
				LogPrint(0, LOG_NETWORK, 0, "Could not start SSLConnection thread...");
				SSL_shutdown(ssl);
				SSL_free(ssl);
				ssl = NULL;
				disconnectInternal(0);
				mutex.leave();
				return false;
			}
		}

		mutex.leave();
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::connect(uint64 addr, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	#ifdef _USE_SSL_
		// Make client SSL connection; create the client context on demand so the
		// verification policy is always applied
		if (!ctx && !init()) {
			return false;
		}
		if (!ctx || !(ssl = SSL_new(ctx))) {
			LogSSLErrors("SSL_new", 0);
			return false;
		}

		// Hostname verification: when peer verification is on and an expected
		// hostname is known (recorded by connect(const char*,...) or set via
		// setVerifyHostName), require the peer certificate to match it and
		// send SNI. With allowSelfSigned or no hostname (raw IP connects),
		// only the chain-of-trust check (or none) applies, as before.
		if (!allowSelfSigned && !verifyHostName.empty()) {
			if (SSL_set1_host(ssl, verifyHostName.c_str()) != 1) {
				LogPrint(0, LOG_NETWORK, 0, "SSL: could not set expected hostname '%s' for verification", verifyHostName.c_str());
				SSL_free(ssl);
				ssl = NULL;
				return false;
			}
			SSL_set_tlsext_host_name(ssl, verifyHostName.c_str());
		}

		mutex.enter(1000);
		if (timeoutMS) {
			if (!connectTimeoutMS)
				connectTimeoutMS = timeoutMS;
		}
		else
			timeoutMS = connectTimeoutMS;
		if((socket=::socket(AF_INET,SOCK_STREAM,IPPROTO_TCP))==INVALID_SOCKET){
			LogPrint(0, LOG_NETWORK, 0, "Could not create SSLListener socket (%d)...", utils::GetLastOSErrorNumber());
			mutex.leave();
			return false;
		}

		bufferContentLen = 0;
		bufferContentPos = 0;

		// Set blocking mode
		utils::SetSocketNonBlockingMode(socket);

		struct sockaddr_in saServer;
		saServer.sin_family = AF_INET;
		saServer.sin_port = htons(GETIPPORT(addr));
		memcpy(&saServer.sin_addr.s_addr, &addr, 4);


		// Connect to the server
		int res;
		if ((res = ::connect(socket, (struct sockaddr*)&saServer, sizeof(struct sockaddr))) != 0) {

			int err = utils::GetLastOSErrorNumber();
			if ( (err == SOCKETWOULDBLOCK) || (err == SOCKETTRYAGAIN) ) {
				if (!didConnect(1000)) {
					SSL_free(ssl);
					ssl = NULL;
					disconnectInternal(0);
					mutex.leave();
					return false;
				}
			}
			else {
				SSL_free(ssl);
				ssl = NULL;
				disconnectInternal(0);
				mutex.leave();
				return false;
			}
		}

		struct linger tmp = {1, 0};
		setsockopt(socket, SOL_SOCKET, SO_LINGER, (char *)&tmp, sizeof(tmp));
		int delay = 1;
		setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, (char*) &delay, sizeof(delay));

		// Connect the SSL struct to our connection
		if (!SSL_set_fd (ssl, (int)socket)) {
			LogSSLErrors("SSL_set_fd", 0);
			SSL_free(ssl);
			ssl = NULL;
			disconnectInternal(0);
			mutex.leave();
			return false;
		}

		//SSL_set_connect_state(ssl);
		SSL_set_mode(ssl, SSL_MODE_AUTO_RETRY);

		// Set blocking mode
		// utils::SetSocketNonBlockingMode(socket);

		int ret; //, err;
		// Initiate SSL handshake
		while ( (ret = SSL_connect(ssl)) != 1) {
			switch (SSL_get_error(ssl, ret)) {
				case SSL_ERROR_WANT_READ:
					utils::WaitForSocketReadability(socket, 200);
					break;
				case SSL_ERROR_WANT_WRITE:
					utils::WaitForSocketWriteability(socket, 200);
					break;
				default:
					// Handshake failed (e.g. peer certificate verification failed).
					// Log at level 1 so the detail is available with network logging
					// on, but stays off the console in quiet runs (incl. the unit
					// test harness's negative-path verify tests).
					LogSSLErrors("handshake failed", 1);
					SSL_free(ssl);
					ssl = NULL;
					disconnectInternal(0);
					mutex.leave();
					return false;
			}
			//err = SSL_get_error(ssl, ret);
			//if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {
			//	utils::WaitForSocketWriteability(socket, 200);
			//	if ( (ret = SSL_connect(ssl)) != 1) {
			//		ERR_print_errors_fp(stderr);
			//		disconnect();
			//		mutex.leave();
			//		return false;
			//	}
			//}
		}

		uint32 certnamemax = 1000;
		char *certname;
		X509 *cert = NULL;

		cert = SSL_get_peer_certificate(ssl);
		if (cert != NULL) {
			certname = new char[certnamemax+1];
			certinfo = X509_NAME_oneline(X509_get_subject_name(cert), certname, certnamemax);
			delete [] certname;
			X509_free(cert);
		}

		if (receiver != NULL) {
			this->receiver = receiver;
			// Start networking thread
			if (!ThreadManager::CreateThread(SSLConnectionRun, this, threadID)) {
				LogPrint(0, LOG_NETWORK, 0, "Could not start SSLConnection thread...");
				SSL_free(ssl);
				ssl = NULL;
				disconnectInternal(0);
				mutex.leave();
				return false;
			}
		}

		remoteAddress = addr;
		remote = false;

		mutex.leave();
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::connect(const char* addr, uint16 port, uint64& location, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	#ifdef _USE_SSL_
		uint32 address;
		if (!utils::LookupIPAddress(addr, address))
			return false;
		// Record the hostname for certificate hostname verification (unless one
		// was set explicitly, or addr is an IP literal)
		if (verifyHostName.empty() && addr && (inet_addr(addr) == INADDR_NONE))
			verifyHostName = addr;
		return connect(location = GETIPADDRESSPORT(address, port), timeoutMS, receiver);
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::connect(const uint32* addresses, uint16 addressCount, uint16 port, uint64& location, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	#ifdef _USE_SSL_
		// Try all addresses
		for (uint16 n=0; n<addressCount; n++) {
			if (connect(location = GETIPADDRESSPORT(addresses[n], port), timeoutMS, receiver))
				return true;
		}
		return false;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::delayedConnect(uint64 addr, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	#ifdef _USE_SSL_
	mutex.enter(1000);

		connectTimeoutMS = timeoutMS;
		remoteAddress = addr;
		remote = false;

		if (receiver != NULL) {
			this->receiver = receiver;
			// Start networking thread
			if (!ThreadManager::CreateThread(SSLConnectionRun, this, threadID)) {
				LogPrint(0, LOG_NETWORK, 0, "Could not start SSLConnection thread...");
				mutex.leave();
				return false;
			}
		}

		mutex.leave();
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::delayedConnect(const char* addr, uint16 port, uint64& location, uint32 timeoutMS, NetworkDataReceiver* receiver) {
	#ifdef _USE_SSL_
		// Resolve address
		uint32 address;
		if (!utils::LookupIPAddress(addr, address))
			return false;
		// Record the hostname for certificate hostname verification (unless one
		// was set explicitly, or addr is an IP literal)
		if (verifyHostName.empty() && addr && (inet_addr(addr) == INADDR_NONE))
			verifyHostName = addr;
		return delayedConnect(location = GETIPADDRESSPORT(address, port), timeoutMS, receiver);
	#else // _USE_SSL_
	return false;
	#endif // _USE_SSL_
}

bool SSLConnection::send(const char* data, uint32 size, uint64 receiver) {
	#ifdef _USE_SSL_
		if (!size)
			return false;
		// Ignore receiver, only for UDP
		if (!sendMutex.enter(500))
			return false;
		if (!ssl) {
			sendMutex.leave();
			return false;
		}
		uint64 start = GetTimeNow();
		//utils::SetSocketBlockingMode(socket);

		int ret;
		while ( (ret = SSL_write(ssl, data, size)) <= 0) {
			switch (SSL_get_error(ssl, ret)) {
				case SSL_ERROR_WANT_READ:
					utils::WaitForSocketReadability(socket, 200);
					break;
				case SSL_ERROR_WANT_WRITE:
					utils::WaitForSocketWriteability(socket, 200);
					break;
				default:
					sendMutex.leave();
					disconnect(NETWORKERROR_RECEIVE);
					return false;
			}
		}

		if (ret < (int)size) {
			sendMutex.leave();
			disconnect(NETWORKERROR_SEND_TIMEOUT);
			return false;
		}
		utils::WaitForSocketWriteability(socket, 1000);
		//utils::SetSocketNonBlockingMode(socket);

		int64 t = GetTimeAge(start);
		if (t > 0)
			outputSpeed = (uint32)(size*1000000.0/(double)t);
		outputBytes += size;
		sendMutex.leave();

		#ifdef TCPCON_PRINT_DEBUG
			if (size < 1024) {
				char* tmp = new char[size+1];
				memcpy(tmp, data, size);
				tmp[size] = 0;
				LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> SSL Sent %u bytes '%s'\n", size, tmp);
				delete [] tmp;
			}
			else {
				#ifdef TCPCON_PRINTBINARY_DEBUG
					char* tmp = new char[size+1];
					memcpy(tmp, data, size);
					tmp[size] = 0;
					LogPrint(0,LOG_NETWORK,0,">>>>>>> SEND >>>>>>>> SSL Sent %u bytes '%s'\n", size, tmp);
					delete [] tmp;
				#endif
			}
		#endif
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}



bool SSLConnection::reconnect(uint32 timeoutMS) {
	#ifdef _USE_SSL_
		disconnect();
		return connect(remoteAddress, timeoutMS, receiver);
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}


bool SSLConnection::findRemoteAddress(uint64& addr) {

	if (socket == INVALID_SOCKET)
		return false;

	struct sockaddr_in remoteAddr;
	
	#ifdef WINDOWS
		int remoteAddrLen;
	#else
		#ifdef Darwin
			#if GCC_VERSION < 40000
				int remoteAddrLen;
			#else
				socklen_t remoteAddrLen;
			#endif // GCC_VERSION < 40000
		#else
			socklen_t remoteAddrLen;
		#endif
	#endif // WINDOWS

	remoteAddrLen = sizeof(struct sockaddr_in);

	if (getpeername(socket, (struct sockaddr*) &remoteAddr, &remoteAddrLen) != 0)
		return false;

	uint32 address = remoteAddr.sin_addr.s_addr;
	if ( address == LOCALHOSTIP ) {
		// Get the actual local ip address, if possible
		utils::GetLocalIPAddress(address);
	}
	addr = GETIPADDRESSPORT(address, (uint16)remoteAddr.sin_port);
	return true;
}

int32 SSLConnection::peekStream() {
	#ifdef _USE_SSL_
	mutex.enter(1000);
		if (!ssl) {
			mutex.leave();
			return -1;
		}
		// and read from the socket
		// int count = ::recvfrom(socket,buffer+bufferContentLen,bufferLen-bufferContentLen,MSG_PEEK,NULL,0);
		int count = SSL_pending(ssl);
		mutex.leave();
		return count;
	#else // _USE_SSL_
		return -1;
	#endif // _USE_SSL_
}

int32 SSLConnection::readIntoBuffer() {

	// Assume that the mutex is locked

	#ifdef _USE_SSL_

		if (!ssl)
			return -1;

		int32 count = 0;
		int pending = SSL_pending(ssl);
		if (pending <= 0)
			return 0;

		// Check for buffer resize
		if ((int32)bufferLen-(int32)bufferContentLen < pending) {
			if ((int32)bufferLen-(int32)bufferContentLen + (int32)bufferContentPos > pending) {
				memmove(buffer, buffer+bufferContentPos, bufferContentLen-bufferContentPos);
				bufferContentLen -= bufferContentPos;
				bufferContentPos = 0;
			}
			else
				resizeBuffer(bufferLen + pending + INITIALBUFFERSIZE);
		}

		// Read pending bytes from the socket
		count = SSL_read(ssl, buffer+bufferContentLen, pending);

		if (count < 0)
			return 0;
		else if (count == 0) {
			disconnectInternal(NETWORKERROR_RECEIVE);
			return -1;
		}

		#ifdef TCPCON_PRINT_DEBUG
			char* tmp = new char[count+1];
			memcpy(tmp, buffer+bufferContentLen, count);
			tmp[count] = 0;
			LogPrint(0,LOG_NETWORK,0,"<<<<<<< SSL READINTOBUFFER <<<<<<<< TCP RECV BUF %u bytes '%s'\n", count, tmp);
			delete [] tmp;
		#endif

		bufferContentLen += count;

		return count;
	#else // _USE_SSL_
		return -1;
	#endif // _USE_SSL_
}

bool SSLConnection::receiveAvailable(char* data, uint32& size, uint32 maxSize, uint32 timeout, bool peek) {

	#ifdef _USE_SSL_
	mutex.enter(1000);

		if (!ssl) {
			mutex.leave();
			return false;
		}

		if (maxSize > bufferLen-bufferContentLen+bufferContentPos) {
			if (bufferContentPos > maxSize) {
				memmove(buffer, buffer+bufferContentPos, bufferContentLen-bufferContentPos);
				bufferContentLen -= bufferContentPos;
				bufferContentPos = 0;
			}
			else
				resizeBuffer((bufferLen+maxSize) * 2);
		}

		uint64 start = GetTimeNow(), timespent;

		int count, err, timeleft;
		// Do we have enough data in the buffer already
		while (bufferContentLen - bufferContentPos < maxSize) {
			// and read from the socket
			count = SSL_read(ssl,buffer+bufferContentLen,bufferLen-bufferContentLen);
			if(count <= 0) {
				err = SSL_get_error(ssl, count);
				if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {
				}
				else {
					mutex.leave();
					disconnect(NETWORKERROR_RECEIVE);
					return false;
				}
			}
			else if (count > 0) {
				#ifdef TCPCON_PRINT_DEBUG
					char* tmp = new char[count+1];
					memcpy(tmp, buffer+bufferContentLen, count);
					tmp[count] = 0;
					LogPrint(0,LOG_NETWORK,0,"<<<<<<< RECEIVE <<<<<<<< TCP RECV AVAIL %u bytes '%s'\n", count, tmp);
					delete [] tmp;
				#endif
				bufferContentLen += count;
				if (bufferContentLen - bufferContentPos >= maxSize)
					break;
			}
			if ((timespent = (GetTimeNow() - start)/1000) >= timeout) {
			//	if (timeout > 0)
			//		printf("**** TCP::receive available %u ***\n", bufferContentLen - bufferContentPos);
				break;
			}
			else {
				mutex.leave();
				timeleft = (int)(timeout-timespent);
				utils::WaitForSocketReadability(socket, timeleft < 50 ? timeleft : 50);
			}
			mutex.enter(1000);
		}

		size = bufferContentLen - bufferContentPos;
		if (size > maxSize)
			size = maxSize;

		inputBytes += size;
		if ( size > 0 ) {
			memcpy(data, buffer+bufferContentPos, size);
			if (!peek) {
				bufferContentPos += size;
				if (bufferContentPos == bufferContentLen)
					bufferContentPos = bufferContentLen = 0;
			}
		}
		mutex.leave();
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}

bool SSLConnection::receive(char* data, uint32 size, uint32 timeout, bool peek) {

	#ifdef _USE_SSL_
	mutex.enter(1000);

		if (!ssl) {
			mutex.leave();
			return false;
		}

		if (size > bufferLen-bufferContentLen+bufferContentPos) {
			if (bufferContentPos > size) {
				memmove(buffer, buffer+bufferContentPos, bufferContentLen-bufferContentPos);
				bufferContentLen -= bufferContentPos;
				bufferContentPos = 0;
			}
			else
				resizeBuffer((bufferLen+size) * 2);
		}

		uint64 start = GetTimeNow();
		int32 timespent;

		int count, err, timeleft;
		// Do we have enough data in the buffer already
		while (bufferContentLen - bufferContentPos < size) {
			if (!ssl) {
				mutex.leave();
				return false;
			}

			// and read from the socket
			
			// int count = ::recvfrom(socket,buffer+bufferContentLen,bufferLen-bufferContentLen,0,NULL,0);
			
			count = SSL_read((SSL *)ssl, buffer+bufferContentLen,size);

			if(count <= 0) {
				err = SSL_get_error(ssl, count);
				if (err == SSL_ERROR_WANT_READ || err == SSL_ERROR_WANT_WRITE) {
				}
				else {
					mutex.leave();
					disconnect(NETWORKERROR_RECEIVE);
					return false;
				}
			}
			else if (count > 0) {
				#ifdef TCPCON_PRINTBINARY_DEBUG
					char* tmp = new char[count+1];
					memcpy(tmp, buffer+bufferContentLen, count);
					tmp[count] = 0;
					LogPrint(0,LOG_NETWORK,0,"<<<<<<<< RECEIVE <<<<<<< TCP RECV %u bytes '%s'", count, tmp);
					delete [] tmp;
				#endif
				bufferContentLen += count;
				if (bufferContentLen - bufferContentPos >= size)
					break;
			}
			if ((timespent = GetTimeAgeMS(start)) >= (int32)timeout) {
			//	printf("**** TCP::receive only got %u out of %u bytes ***\n", bufferContentLen - bufferContentPos, size);
				mutex.leave();
				return false;
			}
			else {
				//utils::Sleep(5);
				//uint64 t = GetTimeNow();
				mutex.leave();
				timeleft = (int)(timeout-timespent);
				utils::WaitForSocketReadability(socket, timeleft < 50 ? timeleft : 50);
				//printf("WaitForRead:      %lu    (%lu)\n", GetTimeAgeMS(t), timeout-timespent);
			}
			mutex.enter(1000);
		}

		int64 t = GetTimeAge(start);
		if (t > 0)
			inputSpeed = (uint32)(size*1000000.0/(double)t);
		inputBytes += size;

		memcpy(data, buffer+bufferContentPos, size);
		if (!peek) {
			bufferContentPos += size;
			if (bufferContentPos == bufferContentLen)
				bufferContentPos = bufferContentLen = 0;
		}
		mutex.leave();
		return true;
	#else // _USE_SSL_
		return false;
	#endif // _USE_SSL_
}










THREAD_RET THREAD_FUNCTION_CALL TCPListenerRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((TCPListener*)arg)->run() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL TCPConnectionRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((TCPConnection*)arg)->run() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL UDPConnectionRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((UDPConnection*)arg)->run() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL SSLConnectionRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((SSLConnection*)arg)->run() ? 0 : 1));
}



bool NetworkTest_TCPServer(uint16 port, uint32 count2) {
	uint32 period = 100000;

	TCPListener* listener = new TCPListener();
	if (!listener->init(port, NOENC)) {
		delete(listener);
		printf("Could not bind to port %u, exiting... \n\n", port);
		return false;
	}

	char* startdata = new char[12], *data;
	uint32 size, count;
	NetworkConnection* con;
	uint32 c = 0;
	while (true) {
		c = 0;
		printf("\n\nListening to port %u for a new connection... ", port);
		while ( (con = listener->acceptConnection(10000)) == NULL) {}

		while (con->isConnected()) {
			// Reading count and size
			if (!con->receive(startdata, 12, 10000)) {
				printf("[%u] Could not receive start data, exiting...\n", c);
				break;
			}
			if ((*(uint32*)startdata) != 123456789) {
				printf("[%u] Start data wrong, exiting...\n", c);
				break;
			}
			size = *(((uint32*)startdata)+1);
			count = *(((uint32*)startdata)+2);
			c = 0;
			data = new char[size];

			printf("Got it - starting test with size %u and count %u...\n\n", size, count);

			while (con->isConnected() && (c++ < count) ) {
				if (!con->receive(data, size, 10000)) {
					printf("[%u] Could not receive data %u, exiting...\n", port, c);
					break;
				}
				if (!con->send(data, size)) {
					printf("[%u] Could not send data %u, exiting...\n", port, c);
					break;
				}
			}
			delete [] data;
		}
		delete(con);


		//while (con->isConnected()) {
		//	t1 = GetTimeNow();
		//	if (!NetworkTest_SendReceiveData(con, data, dataLen, c, true)) {
		//		delete(con);
		//		break;
		//	}
		//	t2 = GetTimeNow();
		//	d += t2-t1;
		//	if ((c > 0) && (c % period == 0)) {
		//		printf("Round time: %.3f us...\n", (double)d/period);
		//		d = 0;
		//	}
		//	c++;
		//	if ( (count > 0) && (c > count) ) {
		//		delete(con);
		//		delete(listener);
		//		return true;;
		//	}
		//}

	}
	delete [] startdata;
	return true;
}


bool NetworkTest_TCPClient(const char* address, uint16 port, uint32 count2) {
	TCPConnection* con = new TCPConnection();
	uint64 location;
	if (!con->connect(address, port, location, 5000)) {
		delete(con);
		printf("Could not connect to '%s' on port %u, exiting...\n", address, port);
		return false;
	}
	printf("Connected to '%s' on port %u, starting test...\n", address, port);

	char* startdata = new char[12];
	memset(startdata, 0, 12);
	*((uint32*)startdata) = 123456789;

	uint32 maxSize = 1024*64, innercount = 20, count = 1000, steps = 64, s, c, step;
	char* data = new char[maxSize];
	double* vals = new double[count];
	double* avgvals = new double[steps];
	double* maxvals = new double[steps];
	double* minvals = new double[steps];
	double* stdvals = new double[steps];
	double sum, avg, mx, mn, std;

	uint64 t1, t2;

	while (con->isConnected()) {
	
		for (step=0; step<steps; step++) {
			s = (maxSize/steps*(step+1));
			*(((uint32*)startdata)+1) = s;
			*(((uint32*)startdata)+2) = count * innercount;
			if (!con->send(startdata, 12)) {
				printf("[%u] Could not send startdata, exiting...\n", port);
				break;
			}
			printf("Step %u size %u...\n", step, s);
			for (c=0; c<count; c++) {
				t1 = GetTimeNow();
				for (uint32 i=0; i<innercount; i++) {
					if (!con->send(data, s)) {
						printf("[%u] Could not send data %u, exiting...\n", port, c);
						break;
					}
					if (!con->receive(data, s, 10000)) {
						printf("[%u] Could not receive data %u, exiting...\n", port, c);
						break;
					}
				}
				t2 = GetTimeNow();
				vals[c] = (t2-t1)/(double)innercount;
			}

			sum = avg = std = 0;
			mx = mn = vals[0];
			for (c=0; c<count; c++) {
				sum += vals[c];
				if (vals[c] > mx) mx = vals[c];
				if (vals[c] < mn) mn = vals[c];
			}
			avg = sum/count;

			sum = 0;
			for (c=0; c<count; c++)
				sum += pow((vals[c] - avg), 2);
			std = sqrt(sum/count-1);

			avgvals[step] = avg;
			maxvals[step] = mx;
			minvals[step] = mn;
			stdvals[step] = std;
		}
	
		printf("Test results (size, avg, min, max, std in us):\n");
		for (step=0; step<steps; step++) {
			s = (maxSize/steps*(step+1));
			printf("%u  %u  %.3f  %.3f  %.3f  %.3f\n",
				step, s, avgvals[step], minvals[step], maxvals[step], stdvals[step]);
		}

		break;
	}

	delete [] startdata;
	delete [] data;
	delete [] vals;
	delete [] avgvals;
	delete [] maxvals;
	delete [] minvals;
	delete [] stdvals;




//	strcpy(data+(2*sizeof(uint32)), "Testing");

//	utils::PrintBinary(data, dataLen, true, "Initial Structure");

//	uint32 c = 0;
//	uint64 t1, t2, d = 0;






	//while (con->isConnected()) {
	//	t1 = GetTimeNow();
	//	if (!NetworkTest_SendReceiveData(con, data, dataLen, c)) {
	//		delete(con);
	//		return false;
	//	}
	//	t2 = GetTimeNow();
	//	d += t2-t1;
	//	if ((c > 0) && (c % period == 0)) {
	//		printf("Round time: %.3f us...\n", (double)d/period);
	//		d = 0;
	//	}
	//	c++;
	//	if ( (count > 0) && (c > count) )
	//		break;
	//}
	delete(con);
	return true;
}

bool NetworkTest_SendReceiveData(TCPConnection* con, char* data, uint32 dataLen, uint32 c, bool receiveFirst) {

	uint32 expectC = c*2;
	if (!receiveFirst) {
		expectC++;
		// Send data
		if (!con->send(data, dataLen)) {
			printf("[%u] Could not send data, exiting...\n", c);
			return false;
		}
		//utils::PrintBinary(data, dataLen, true, "Send1");
	}

	memset(data, 0, dataLen);
	// Wait for reply data
	if (!con->receive(data, dataLen, 10000)) {
		printf("[%u] Could not receive data, exiting...\n", c);
		return false;
	}		
	//utils::PrintBinary(data, dataLen, true, "Receive");
	// Check reply data
	if (*(uint32*)data != dataLen) {
		printf("[%u] Did not receive correct length (%u != %u), exiting...\n", c, *(uint32*)data, dataLen);
	//	utils::PrintBinary(data, dataLen, true, "Structure");
		return false;
	}		
	// Check reply data
	if (*((uint32*)data+1) != expectC) {
		printf("[%u] Did not receive correct count (%u != %u), exiting...\n", c, *((uint32*)data+1), expectC);
	//	utils::PrintBinary(data, dataLen, true, "Structure");
		return false;
	}		
	// Check reply data
	if (strcmp(data+(2*sizeof(uint32)), "Testing") != 0) {
		data[dataLen-1] = 0;
		printf("[%u] Did not receive correct text ('%s' != 'Testing'), exiting...\n", c, data+(2*sizeof(uint32)));
	//	utils::PrintBinary(data, dataLen, true, "Structure");
		return false;
	}
	*((uint32*)data+1) = *((uint32*)data+1) + 1;

//	printf("[%u] Receive OK\n", c);
	if (receiveFirst) {
		// Send data
		if (!con->send(data, dataLen)) {
			printf("[%u] Could not send data, exiting...\n", c);
			return false;
		}
		//utils::PrintBinary(data, dataLen, true, "Send2");
	}
	return true;
}


}
