

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 RequestClient.cpp
	@brief Implementation of the request-system client side: RequestReply
	completion/waiting, RequestClient's gateway connection upkeep, request
	dispatch and reply matching, plus the load-test clients used by the
	gateway unit test.
	@ingroup Requests */
#include "RequestClient.h"
#include "UnitTestFramework.h"

namespace	cmlabs{


/////////////////////////////////////////////////////////////
// Request Connection
/////////////////////////////////////////////////////////////

uint32 RequestConnection::removeStaleRequests(uint32 ttlMS) {
	uint32 count = longReqQueue.removeStaleRequests(ttlMS);
	count += shortReqQueue.removeStaleRequests(ttlMS);
	return count;
}


/////////////////////////////////////////////////////////////
// Request Reply
/////////////////////////////////////////////////////////////

bool RequestReply::isComplete() {
	switch (this->status) {
		case SUCCESS:
		case FAILED:
		case TIMEOUT:
		case TOOBUSY:
		case LOCALERROR:
		case NETWORKERROR:
		case SERVERERROR:
			return true;
		default:
			return false;
	}
	return false;
}

bool RequestReply::setStatus(RequestStatus status) {
	if (!mutex.enter(1000))
		return false;
	this->status = status;
	mutex.leave();
	return true;
}

RequestStatus RequestReply::getStatus() {
	RequestStatus res = LOCALERROR;
	if (!mutex.enter(1000))
		return res;
	res = this->status;
	mutex.leave();
	return res;
}


std::string RequestReply::getStatusText() {
	RequestStatus res = LOCALERROR;
	if (!mutex.enter(1000))
		return "Locking error";
	res = this->status;
	mutex.leave();
	switch (res) {
	case IDLE:
		return "Request idle";
	case QUEUED:
		return "Request queued";
	case SENT:
		return "Request sent";
	case SUCCESS:
		return "Request success";
	case FAILED:
		return "Request failed";
	case TIMEOUT:
		return "Request timed out";
	case LOCALERROR:
		return "Request local error";
	case NETWORKERROR:
		return "Request network error";
	case SERVERERROR:
		return "Request server error";
	case TOOBUSY:
		return "Request server too busy";
	default:
		return "Error getting status";
	}
}

RequestCallbackFunction RequestReply::getCallback() {
	RequestCallbackFunction res = NULL;
	if (!mutex.enter(1000))
		return res;
	res = this->callback;
	mutex.leave();
	return res;
}

bool RequestReply::setCallback(RequestCallbackFunction callback) {
	if (!mutex.enter(1000))
		return false;
	this->callback = callback;
	mutex.leave();
	return true;
}

bool RequestReply::giveRequestMessage(DataMessage* msg) {
	if (!mutex.enter(1000))
		return false;
	if (requestMsg)
		delete(requestMsg);
	requestMsg = msg;
	systemID = msg->getSystemID();
	mutex.leave();
	return true;
}

bool RequestReply::setRequestMessageCopy(DataMessage* msg) {
	if (!mutex.enter(1000))
		return false;
	if (requestMsg)
		delete(requestMsg);
	if (msg)
		requestMsg = new DataMessage(*msg);
	else
		requestMsg = NULL;
	systemID = msg->getSystemID();
	mutex.leave();
	return true;
}

DataMessage* RequestReply::getRequestMessageCopy() {
	DataMessage* msg = NULL;
	if (!mutex.enter(1000))
		return NULL;
	msg = new DataMessage(*requestMsg);
	mutex.leave();
	return msg;
}

DataMessage* RequestReply::peekRequestMessage() {
	return requestMsg;
}

bool RequestReply::giveReplyMessage(DataMessage* msg) {
	if (!mutex.enter(1000))
		return false;
	if (replyMsg)
		delete(replyMsg);
	replyMsg = msg;
	mutex.leave();
	return true;
}

bool RequestReply::setReplyMessageCopy(DataMessage* msg) {
	if (!mutex.enter(1000))
		return false;
	if (replyMsg)
		delete(replyMsg);
	if (msg)
		replyMsg = new DataMessage(*msg);
	else
		replyMsg = NULL;
	mutex.leave();
	return true;
}

DataMessage* RequestReply::getReplyMessageCopy() {
	DataMessage* msg = NULL;
	if (!mutex.enter(1000))
		return NULL;
	msg = new DataMessage(*replyMsg);
	mutex.leave();
	return msg;
}

DataMessage* RequestReply::peekReplyMessage() {
	return replyMsg;
}

bool RequestReply::replyToRequest(DataMessage* msg, RequestStatus status) {
	//if (!msg) return false;
	if (!mutex.enter(1000))
		return false;
	if (replyMsg)
		delete(replyMsg);
	replyMsg = msg;
	this->status = status;
	finishTime = GetTimeNow();
	semaphore.signal();

	//if (customRef == 2) {
	//	printf("--- [%u] --- Long request %.3fms \n", msg->getFrom(), ((double)finishTime-startTime)/1000.0);
	//	fflush(stdout);
	//}
	//else if (customRef == 1) {
	//	printf("--- [%u] --- Short request %.3fms \n", msg->getFrom(), ((double)finishTime-startTime)/1000.0);
	//	fflush(stdout);
	//}

	mutex.leave();
	return true;
}

uint64 RequestReply::getRequestDuration() {
	if (finishTime)
		return finishTime-startTime;
	else
		return 0;
}

uint32 RequestReply::getRequestDurationMS() {
	if (finishTime)
		return (uint32)((finishTime-startTime)/1000);
	else
		return 0;
}

RequestStatus RequestReply::waitForResult(uint32 timeoutMS) {
	RequestStatus result;
	if (!mutex.enter(1000, __FUNCTION__))
		return LOCALERROR;

	//if ( (status == SUCCESS) || (status == FAILED) || (status == TIMEOUT) || (status == LOCALERROR) || (status == NETWORKERROR) || (status == SERVERERROR) ) {
	if (isComplete()) {
		result = status;
		mutex.leave();
		return result;
	}

	mutex.leave();
	if (semaphore.wait(timeoutMS)) {
		if (!mutex.enter(1000, __FUNCTION__))
			return LOCALERROR;
		//if ( (status == SUCCESS) || (status == FAILED) || (status == TIMEOUT) || (status == LOCALERROR) || (status == NETWORKERROR) || (status == SERVERERROR) ) {
		if (isComplete()) {
			result = status;
			mutex.leave();
			return result;
		}
		mutex.leave();
	}
	return TIMEOUT;
}

DataMessage* RequestReply::waitForMessage(uint32 timeoutMS, bool takeMessage) {
	// Fast path: if the reply (or a terminal status) is already in, return it
	// under the mutex. Otherwise block on the semaphore that replyToRequest()
	// signals, then re-check under the mutex — the double-check is needed
	// because the semaphore can also be signalled for failure statuses where
	// replyMsg stays NULL. With takeMessage the internal pointer is cleared so
	// ownership moves to the caller (and finishRequest() won't double-delete).
	DataMessage* msg;
	if (!mutex.enter(1000, __FUNCTION__))
		return NULL;

	if (replyMsg || finishTime) {
		msg = replyMsg;
		if (takeMessage)
			replyMsg = NULL;
		mutex.leave();
		return msg;
	}

	mutex.leave();
	if (semaphore.wait(timeoutMS)) {
		if (!mutex.enter(1000, __FUNCTION__))
			return NULL;
		if (replyMsg) {
			msg = replyMsg;
			if (takeMessage)
				replyMsg = NULL;
			mutex.leave();
			return msg;
		}
		mutex.leave();
	}
	return NULL;
}


/////////////////////////////////////////////////////////////
// Request Client
/////////////////////////////////////////////////////////////

RequestClient::RequestClient() {
	lastRefID = 0;
	sentCount = 0;
	receivedCount = 0;
	shouldContinue = true;
	isRunning = false;
	channel = NULL;
	manager = new NetworkManager();

	if (!ThreadManager::CreateThread(RequestClientRun, this, threadID, 0)) {
		shouldContinue = false;
	}
}

RequestClient::~RequestClient() {
	conMutex.enter(3000, __FUNCTION__);
	stop();

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

	delete(manager);
}

bool RequestClient::addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption) {
	
	std::list<RequestGatewayConnection>::iterator i, e = connections.end();
	if (!mutex.enter(1000, __FUNCTION__))
		return false;
	// Check to see if this gateway is already in the list
	i = connections.begin();
	while (i != e) {
		if ( (stricmp((*i).addr.c_str(), addr.c_str()) == 0) && ((*i).port == port) ) {
			mutex.leave();
			return true;
		}
		i++;
	}

	RequestGatewayConnection con;
	con.clear();
	con.id = id;
	con.addr = addr;
	con.port = port;
	con.encryption = encryption;

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

	DataMessage* msgConnect = new DataMessage();
	msgConnect->setString("URI", "ClientConnect");

	if (!channel)
		channel = manager->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, 0, this, con.conID, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());
	else
		con.conID = channel->addTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, con.location, 1000, (const char*)msgConnect->data, msgConnect->getSize());

	delete msgConnect;

	connections.push_back(con);
	mutex.leave();
	return true;
}

RequestReply* RequestClient::postRequest(DataMessage *msg) {
	// Correlation scheme: each outgoing request gets the next monotonically
	// increasing clientRef, stamped into the message's reference field. The
	// gateway echoes it back in the reply, and receiveMessage() uses it to
	// find this RequestReply in requestMap. The message itself is owned by
	// the reply object from here on; the worker thread (run()) picks the
	// reply up from outQ and dispatches it to a connected gateway.
	if (!msg)
		return NULL;
	if (!mutex.enter(1000, __FUNCTION__))
		return NULL;
	msg->setReference(++lastRefID);
	
	RequestReply* reply = new RequestReply();
	reply->setStatus(QUEUED);
	reply->giveRequestMessage(msg);
	reply->clientRef = lastRefID;
	requestMap[lastRefID] = reply;
	outQ.add(reply);
	mutex.leave();
	return reply;
}

bool RequestClient::postRequest(DataMessage *msg, RequestCallbackFunction callback, uint32 timeoutMS) {
	// ##############
	return false;
}

bool RequestClient::finishRequest(RequestReply* reply) {
	// Deletes the RequestReply found in requestMap — and with it both the
	// request and (unless taken via waitForMessage(..., true)) the reply
	// DataMessage. Despite the isInUse pooling flag, objects are currently
	// deleted outright rather than recycled (see the commented-out branch).
	if (!reply || !reply->clientRef) return false;
	if (!mutex.enter(1000, __FUNCTION__))
		return false;

	std::map<uint64, RequestReply*>::iterator i = requestMap.find(reply->clientRef);
	if (i == requestMap.end())
		return false;
	if (i->second) {
		delete i->second;
		requestMap.erase(i);
		//if (i->second->isComplete()) {
		//	delete i->second;
		//	requestMap.erase(i);
		//}
		//else
		//	i->second->isInUse = false;
	}
	mutex.leave();
	return true;
}


bool RequestClient::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
	uint64 now = GetTimeNow();
	std::list<RequestGatewayConnection>::iterator i, e = connections.end();
	if (!mutex.enter(1000, __FUNCTION__))
		return false;
	i = connections.begin();
	while (i != e) {
		if ((*i).conID == conid) {
			switch (evt->type) {
			case NETWORKEVENT_DISCONNECT:
			case NETWORKEVENT_DISCONNECT_RETRYING:
				LogPrint(0, LOG_SYSTEM, 0, "Disconnected from gateway %u on %s:%u", conid, (*i).addr.c_str(), (*i).port);
				(*i).lastFailTime = now;
				(*i).lastConTime = 0;
				//(*i).conID = 0;
				break;
			case NETWORKEVENT_CONNECT:
			case NETWORKEVENT_RECONNECT:
				LogPrint(0, LOG_SYSTEM, 0, "Connected to gateway %u on %s:%u", conid, (*i).addr.c_str(), (*i).port);
				(*i).lastFailTime = 0;
				(*i).lastConTime = now;
				break;
			default:
				break;
			}
		}
		i++;
	}
	delete evt;
	mutex.leave();
	return true;
}

bool RequestClient::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
	// Network-thread reply path: look up the pending RequestReply by the
	// echoed reference id and complete it via replyToRequest(), which takes
	// ownership of msg, wakes any waitForResult()/waitForMessage() waiter and
	// fires the callback. The terminal status comes from the message's own
	// status field (set by the gateway: SUCCESS, TIMEOUT, TOOBUSY, ...).
	// Unmatched replies (already finished/timed out locally) are dropped.
	if (!msg) return false;
	if (!mutex.enter(1000, __FUNCTION__)) {
		LogPrint(0, LOG_SYSTEM, 0, "Error locking Client mutex");
		delete(msg);
		return true;
	}
	uint64 ref = msg->getReference();
	std::map<uint64,RequestReply*>::iterator i = requestMap.find(ref);

	if (i == requestMap.end()) {
		// #############
		LogPrint(0, LOG_SYSTEM, 0, "Couldn't find reply ID %llu", ref);
		mutex.leave();
		delete(msg);
		return true;
	}
	RequestReply* reply = i->second;
	//reply->replyToRequest(msg, SUCCESS);
	reply->replyToRequest(msg, (RequestStatus)msg->getStatus());
	if (!reply->isInUse) {
		LogPrint(0, LOG_SYSTEM, 0, "Reply ID %llu no longer in use", ref);
		requestMap.erase(i);
		delete reply;
	}
	//else
	//	LogPrint(0, LOG_SYSTEM, 0, "Got reply ID %llu - status %u", ref, msg->getStatus());
	receivedCount++;
	mutex.leave();
	return true;
}

bool RequestClient::receiveHTTPReply(HTTPReply* reply, HTTPRequest* req, NetworkChannel* channel, uint64 conid) {
	return false;
}


bool RequestClient::sendMessageToGateway(DataMessage* msg, RequestGatewayConnection& con) {

	uint64 now = GetTimeNow();
	if (!channel || !con.lastConTime)
		return false;
	LogPrint(0, LOG_NETWORK, 3, "Sending data to gateway %u on %s:%u...", con.id, con.addr.c_str(), con.port);

	if (!channel->sendMessage(msg, con.conID)) {
		if (!con.lastFailTime)
			LogPrint(0, LOG_SYSTEM, 0, "Unable to send data to gateway %u on %s:%u, disconnecting...", con.id, con.addr.c_str(), con.port);
		con.lastFailTime = GetTimeNow();
		// clearly not working, disconnect and try again later
		//channel->endConnection(con.conID);
		//con.conID = 0;
		return false;
	}
	return true;

	//if (!con.conID || !channel) {
	//	if (!channel)
	//		channel = manager->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.id, this, con.conID, con.location);
	//	else
	//		con.conID = channel->createTCPConnection(con.addr.c_str(), con.port, con.encryption, PROTOCOL_MESSAGE, true, true, con.location);
	//	if (!channel || !con.conID) {
	//		con.lastFailTime = GetTimeNow();
	//		return false;
	//	}
	//	DataMessage* msg = new DataMessage();
	//	msg->setString("URI", "ClientConnect");
	//	if (!channel->sendMessage(msg, con.conID)) {
	//		// clearly not working, disconnect and try again later
	//		channel->endConnection(con.conID);
	//		con.conID = 0;
	//		con.lastFailTime = GetTimeNow();
	//		return false;
	//	}
	//	con.lastConTime = GetTimeNow();
	//}
	//if (!channel->sendMessage(msg, con.conID)) {
	//	// clearly not working, disconnect and try again later
	//	channel->endConnection(con.conID);
	//	con.conID = 0;
	//	con.lastFailTime = GetTimeNow();
	//	return false;
	//}
	//con.lastFailTime = 0;
	//return true;
}


bool RequestClient::run() {
	// Worker loop: (1) every ~3s push a ClientStatus heartbeat to all gateways
	// (also doubles as a liveness probe that triggers reconnection inside
	// sendMessageToGateway); (2) drain outQ with a 50ms blocking wait and hand
	// each pending RequestReply to sendRequest(), which picks a connected
	// gateway. Reply matching happens on the network thread in
	// receiveMessage(), not here.
	RequestReply* reply;
	isRunning = true;

	// should regularly test for old replies to be deleted...
	// ###############

	int32 heartbeatInterval = 3000;
	uint64 lastHeartbeat = 0;

	while (shouldContinue) {
		if (!lastHeartbeat || GetTimeAgeMS(lastHeartbeat) > heartbeatInterval) {
			sendStatusNow();
			lastHeartbeat = GetTimeNow();
		}

		////if (!mutex.enter(200, __FUNCTION__))
		////	continue;
		//if (!channel && connections.size()) {
		//	conMutex.enter(3000, __FUNCTION__);
		//	DataMessage* msg = new DataMessage();
		//	msg->setString("URI", "ClientStatus");
		//	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
		//	iCon = connections.begin();
		//	while (iCon != eCon) {
		//		if (sendMessageToGateway(msg, (*iCon)))
		//			break;
		//		iCon++;
		//	}
		//	delete msg;
		//	conMutex.leave();
		//}

		if (reply = outQ.waitForAndTakeFirst(50))
			sendRequest(reply);

		////mutex.leave();
	}
	isRunning = false;
	return true;
}

uint32 RequestClient::sendStatusNow() {

	if (!conMutex.enter(1000, __FUNCTION__)) {
		LogPrint(0, LOG_SYSTEM, 0, "Mutex error sending heartbeat from Client...");
		return 0;
	}
	if (!connections.size()) {
		//LogPrint(0, LOG_SYSTEM, 0, "No connections sending heartbeat from Executor %u...", executorID);
		conMutex.leave();
		return 0;
	}

	DataMessage* msg = new DataMessage();
	msg->setString("URI", "ClientStatus");

	uint32 success = 0, failed = 0, notConnected = 0;

	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		if (sendMessageToGateway(msg, (*iCon)))
			success++;
		else
			failed++;
		iCon++;
	}
	conMutex.leave();

	delete(msg);
	if (failed)
		LogPrint(0, LOG_SYSTEM, 2, "Failed sending heartbeat from Client to %u gateways (%u succeeded)...", failed, success);
	//	else
	//		LogPrint(0, LOG_SYSTEM, 0, "Sent heartbeat from Executor %u to %u gateways...", executorID, success);
	return success;
}



bool RequestClient::waitForConnection(uint32 timeoutMS) {
	bool isCon = false;
	uint64 start = GetTimeNow();
	while (!(isCon = isConnected()) && (GetTimeAgeMS(start) < (int32)timeoutMS))
		utils::Sleep(50);
	return isCon;
}

bool RequestClient::isConnected() {
	if (!conMutex.enter(3000, __FUNCTION__))
		return false;
	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		//printf("RUN0\n");fflush(stdout);
		//if ((*iCon).conID && channel->isConnected((*iCon).conID)) {
		if ((*iCon).conID && (*iCon).lastConTime) {
			conMutex.leave();
			return true;
		}
		iCon++;
	}
	conMutex.leave();
	return false;
	//bool res = false;
	//// None were connected, try to connect
	//DataMessage* msg = new DataMessage();
	//msg->setString("URI", "ClientStatus");
	//iCon = connections.begin();
	//while (iCon != eCon) {
	//	if (sendMessageToGateway(msg, (*iCon)))
	//		break;
	//	iCon++;
	//}
	//conMutex.leave();
	//return res;
}

bool RequestClient::reconnect() {
	if (!conMutex.enter(3000, __FUNCTION__))
		return false;
	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		if ((*iCon).conID && channel->isConnected((*iCon).conID)) {
			channel->endConnection((*iCon).conID);
		}
		iCon++;
	}
	conMutex.leave();
	return true;
}


bool RequestClient::sendRequest(RequestReply* reply) {
	if (!reply)
		return false;

	uint64 now = GetTimeNow();
	conMutex.enter(3000, __FUNCTION__);
	std::list<RequestGatewayConnection>::iterator iCon, eCon = connections.end();
	iCon = connections.begin();
	while (iCon != eCon) {
		if (sendMessageToGateway(reply->peekRequestMessage(), (*iCon))) {
			reply->setStatus(SENT);
			sentCount++;
			conMutex.leave();
			return true;
		}
		iCon++;
	}
	// We didn't manage to send it, mark it as failed
	reply->replyToRequest(NULL, NETWORKERROR);
	//reply->setStatus(NETWORKERROR);
	conMutex.leave();
	return false;
}

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






std::string RequestClient::GetJSONReplyParameter(const char* json, const char* name) {

	jsmn_parser parser;
	jsmntok_t* tokens = new jsmntok_t[1024];
	int tokenCount = 0;
	const char* val;
	int valSize;
	jsmn_init(&parser);
	if ((tokenCount = jsmn_parse(&parser, json, strlen(json), tokens, 1024)) <= 0) {
		// No JSON found
		delete[] tokens;
		return "";
	}
	if (!(val = GetJSONValue(tokens, tokenCount, json, name, JSMN_UNDEFINED, valSize))) {
		// JSON key not found
		delete[] tokens;
		return "";
	}
	delete[] tokens;
	std::string value = std::string(val, valSize);
	return value.c_str();
}

DataMessage* RequestClient::SendRequestAndWaitForReply(RequestClient* client, DataMessage* msg, int timeoutMS) {
	//if (!client->isConnected())
	//	return NULL;

	DataMessage *replyMsg;
	RequestReply* reply;
	if (reply = client->postRequest(msg)) {
		if (replyMsg = reply->waitForMessage(timeoutMS, true)) {
			client->finishRequest(reply);
			//delete(reply);
			return replyMsg;
		}
		client->finishRequest(reply);
		//delete(reply);
	}
	return NULL;
}

RequestReply* RequestClient::SendRequestAndWaitForReplyObject(RequestClient* client, DataMessage* msg, int timeoutMS) {
	//DataMessage *replyMsg;
	RequestReply* reply;
	if (reply = client->postRequest(msg)) {
		reply->waitForResult(timeoutMS);
		return reply;
		//client->finishRequest(reply);
	}
	return NULL;
}


std::string RequestClient::SendRequestAndWaitForJSON(RequestClient* client, DataMessage* msg, int timeoutMS) {
	//if (!client->isConnected())
	//	return "";

	DataMessage *replyMsg = SendRequestAndWaitForReply(client, msg, timeoutMS);
	if (!replyMsg)
		return "";
	const char* json = replyMsg->getString("JSON");
	if (!json) {
		delete(replyMsg);
		return "";
	}

	std::string jsonReply = json;
	delete(replyMsg);
	return jsonReply;
}

DataMessage* RequestClient::CreateRequestMessage(uint8 operation, const char* req,
	const char* key1, const char* val1,
	const char* key2, const char* val2,
	const char* key3, const char* val3,
	const char* key4, const char* val4,
	const char* key5, const char* val5,
	const char* key6, const char* val6,
	const char* key7, const char* val7,
	const char* key8, const char* val8,
	const char* key9, const char* val9,
	const char* key10, const char* val10,
	const char* key11, const char* val11,
	const char* key12, const char* val12,
	const char* key13, const char* val13,
	const char* key14, const char* val14,
	const char* key15, const char* val15,
	const char* key16, const char* val16,
	const char* key17, const char* val17,
	const char* key18, const char* val18,
	const char* key19, const char* val19,
	const char* key20, const char* val20
	) {
	DataMessage* msg = new DataMessage();
	msg->setInt("HTTP_OPERATION", operation);
	msg->setString("URI", req);
	if (key1) msg->setString(key1, val1);
	if (key2) msg->setString(key2, val2);
	if (key3) msg->setString(key3, val3);
	if (key4) msg->setString(key4, val4);
	if (key5) msg->setString(key5, val5);
	if (key6) msg->setString(key6, val6);
	if (key7) msg->setString(key7, val7);
	if (key8) msg->setString(key8, val8);
	if (key9) msg->setString(key9, val9);
	if (key10) msg->setString(key10, val10);
	if (key11) msg->setString(key11, val11);
	if (key12) msg->setString(key12, val12);
	if (key13) msg->setString(key13, val13);
	if (key14) msg->setString(key14, val14);
	if (key15) msg->setString(key15, val15);
	if (key16) msg->setString(key16, val16);
	if (key17) msg->setString(key17, val17);
	if (key18) msg->setString(key18, val18);
	if (key19) msg->setString(key19, val19);
	if (key20) msg->setString(key20, val20);
	return msg;
}





bool RequestClient::TestServerLogin(const char* address, uint16 port, const char* username, const char* password, const char* reqAfterLogin) {

	RequestClient* reqClient = new RequestClient();
	//	api->logPrint(1, "Connecting to Face Server on %s:%u...", addr.c_str(), port, username.c_str());
	if (!reqClient->addGateway(1, address, port, NOENC)) {
		LogPrint(0, 0, 1, "Could not connect to Face Server on %s:%u...", address, port);
		delete(reqClient);
		return false;
	}

	std::string json;
	if (!(json = RequestClient::SendRequestAndWaitForJSON(
		reqClient,
		RequestClient::CreateRequestMessage(HTTP_POST, "authenticate", "email", username, "password", password, "clientid", "1", "locationid", "1"),
		5000)).length()) {
		LogPrint(0, 0, 1, "Couldn't login to server account...");
		return false;
	}
	std::string token = RequestClient::GetJSONReplyParameter(json.c_str(), "token");
	if (!token.length()) {
		LogPrint(0, 0, 1, "Client login failed: %s...", RequestClient::GetJSONReplyParameter(json.c_str(), "text").c_str());
		return false;
	}

	LogPrint(0, 0, 1, "Successfully logged in with token: %s...", token.c_str());

	if (!reqClient->isConnected()) {
		LogPrint(0, 0, 1, "Couldn't connect to server...");
		return false;
	}

	return true;
}



















TestRequestClient::TestRequestClient(uint32 id) {
	this->id = id;
	port = 10000;
	gatewayCount = 1;
	sendfreq = 1;
	reconnectMS = 0;
	payload = 1024;
	shouldContinue = true;
	isRunning = true;
	shouldFinishUp = false;
}

bool TestRequestClient::init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName) {
	uint32 myThreadID;
	this->port = port;
	if (gatewayCount)
		this->gatewayCount = gatewayCount;
	else
		this->gatewayCount = 1;
	this->sendfreq = sendfreq;
	this->reconnectMS = reconnectMS;
	this->payload = payload;
	this->longTestName = longTestName;
	return ThreadManager::CreateThread(TestRequestClientRun, this, myThreadID, 0);
}

bool TestRequestClient::finishUp() {
	LogPrint(0, LOG_SYSTEM, 0, "Client %u has been asked to finish up...", id);
	shouldFinishUp = true;
	return true;
}

bool TestRequestClient::run() {
	RequestClient* client = NULL;
	//DataMessage* replyMsg;
	RequestReply* reply;
	std::list<RequestReply*> replyList;
	std::list<RequestReply*>::iterator i, e = replyList.end();
	RequestStatus status;

	int32 intervalMS = 1000;
	if (sendfreq > 0)
		intervalMS = (int32)(1000.0 / sendfreq);

	uint64 lastPost = 0;
	uint64 lastReconnect = 0;
	uint64 lastPrint = GetTimeNow();
	int32 printIntervalMS = 3000;

	uint32 stillWaiting, failed, timeout, errors, success, total, tooBusy, serverError;
	stillWaiting = failed = timeout = errors = success = total = tooBusy = serverError = 0;

	char* bigdata = new char[payload];
	memset(bigdata, 1, payload);

	DataMessage* msgTemplate = new DataMessage();
	msgTemplate->setString("URI", "longtest");
	msgTemplate->setData("BigData", bigdata, payload);

	delete[] bigdata;

	Stats responseTimes;

	while (shouldContinue) {

		if (!client) {
			client = new RequestClient();
			for (uint16 m = 0; m<gatewayCount; m++) {
				if (!client->addGateway(m, "localhost", port + m, NOENC)) {
					LogPrint(0, LOG_SYSTEM, 0, "Couldn't add gateway %u to client %u...\n", m);
					delete msgTemplate;
					return false;
				}
			}
			lastReconnect = GetTimeNow();
			utils::Sleep(100);
		}

		// is it time to send the next message?
		if (!shouldFinishUp && (!lastPost || (GetTimeAgeMS(lastPost) >= intervalMS))) {

			if (reconnectMS < 0) {
				reply = RequestClient::SendRequestAndWaitForReplyObject(client, msgTemplate->copy(), 5000);
				if (!reply) {
					LogPrint(0, 0, 1, "Couldn't send request to server...");
					//delete msgTemplate;
					//return false;
					failed++;
				}

				status = reply->getStatus();
				switch (status) {
				case SUCCESS:
					responseTimes.add((double)reply->getRequestDuration());
					success++;
					break;
				case FAILED:
					failed++;
					break;
				case TOOBUSY:
					tooBusy++;
					break;
				case LOCALERROR:
				case NETWORKERROR:
				case SERVERERROR:
					serverError++;
					break;
				case TIMEOUT:
					timeout++;
					break;
				default:
					stillWaiting++;
					break;
				}
				client->finishRequest(reply);
				delete client;
				client = NULL;
				lastPost = GetTimeNow();
				continue;
			}

			// post request
			reply = client->postRequest(msgTemplate->copy());
			if (!reply) {
				LogPrint(0, 0, 1, "Couldn't post request to server...");
				//delete msgTemplate;
				//return false;
			}
			replyList.push_back(reply);
			lastPost = GetTimeNow();
			total++;
		}

		stillWaiting = 0;
		i = replyList.begin();
		while (i != e) {
			reply = *i;
			status = reply->getStatus();
			switch (status) {
			case SUCCESS:
				responseTimes.add((double)reply->getRequestDuration());
				success++;
				break;
			case FAILED:
				failed++;
				break;
			case TOOBUSY:
				tooBusy++;
				break;
			case LOCALERROR:
			case NETWORKERROR:
			case SERVERERROR:
				serverError++;
				break;
			case TIMEOUT:
				timeout++;
				break;
			default:
				stillWaiting++;
				break;
			}
			if (reply->isComplete()) {
				client->finishRequest(reply);
				i = replyList.erase(i);
			}
			else
				i++;

			/*
			if (!stillWaiting || GetTimeAgeMS(startTime) > 60000)
				break;
			//if (stillWaiting != total) {
			LogPrint(0, LOG_SYSTEM, 0, "----- [%u/%u] success: %u  failed: %u  timeout: %u  errors: %u\n",
				stillWaiting, total, success, failed, timeout, errors);
			//printf("--- [%u] --- Long request %.3fms \n", reply->peekReplyMessage()->getFrom(), reply->getRequestDuration()/1000.0);
			//}
			utils::Sleep(200);
			*/
		}

		if (shouldFinishUp && !stillWaiting) {
			break;
		}
		
		if (GetTimeAgeMS(lastPrint) > printIntervalMS) {
			if (tooBusy || serverError || failed || timeout || errors)
				LogPrint(0, LOG_SYSTEM, 0, "%s[%u] success: %u of %u (w: %u): %.3fms - busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
					shouldFinishUp ? "Finishing " : "",
					id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0, tooBusy, serverError, failed, timeout, errors);
			else
				LogPrint(0, LOG_SYSTEM, 0, "%s[%u] success: %u of %u (w: %u): %.3fms\n",
					shouldFinishUp ? "Finishing " : "",
					id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0);
			lastPrint = GetTimeNow();
		}


		if ((reconnectMS > 0) && (GetTimeAgeMS(lastReconnect) >= reconnectMS)) {
			LogPrint(0, LOG_SYSTEM, 0, "Client %u forcefully reconnecting...");
			delete client;
			client = NULL;
			// delete all unprocessed replies
			replyList.clear();
			continue;
		}

		utils::Sleep(20);
	}

	if (tooBusy || serverError || failed || timeout || errors)
		LogPrint(0, LOG_SYSTEM, 0, "Completed [%u] success: %u of %u (w: %u): %.3fms - busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
			id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0, tooBusy, serverError, failed, timeout, errors);
	else
		LogPrint(0, LOG_SYSTEM, 0, "Completed [%u] success: %u of %u (w: %u): %.3fms\n",
			id, success, total, stillWaiting, responseTimes.getAverage() / 1000.0);

	delete client;
	client = NULL;
	isRunning = false;
	delete msgTemplate;
	return true;
}

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













TestWebRequestClient::TestWebRequestClient(uint32 id) {
	this->id = id;
	port = 10000;
	gatewayCount = 1;
	sendfreq = 1;
	reconnectMS = 0;
	payload = 1024;
	shouldContinue = true;
	isRunning = true;
	shouldFinishUp = false;
}

bool TestWebRequestClient::init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName) {
	uint32 myThreadID;
	this->port = port;
	if (gatewayCount)
		this->gatewayCount = gatewayCount;
	else
		this->gatewayCount = 1;
	this->sendfreq = sendfreq;
	this->reconnectMS = reconnectMS;
	this->payload = payload;
	this->longTestName = longTestName;
	return ThreadManager::CreateThread(TestWebRequestClientRun, this, myThreadID, 0);
}

bool TestWebRequestClient::finishUp() {
	LogPrint(0, LOG_SYSTEM, 0, "Client %u has been asked to finish up...", id);
	shouldFinishUp = true;
	return true;
}

bool TestWebRequestClient::run() {

	int32 intervalMS = 1000;
	if (sendfreq > 0)
		intervalMS = (int32)(1000.0 / sendfreq);
	uint64 lastPost = 0;
	uint64 lastPrint = GetTimeNow();
	int32 printIntervalMS = 3000;

	uint32 stillWaiting, failed, timeout, errors, success, total, tooBusy, serverError;
	stillWaiting = failed = timeout = errors = success = total = tooBusy = serverError = 0;

	NetworkManager* manager = new NetworkManager();

	int8 encryption = NOENC;

	char* bigdata = new char[payload];
	memset(bigdata, 1, payload);

	std::string uriString = utils::StringFormat("/api/longtest");

	HTTPRequest* reqTemplate = new HTTPRequest();
	reqTemplate->createRequest(HTTP_POST, "", uriString.c_str(), bigdata, payload, false, 0);
	delete[] bigdata;

	HTTPRequest* req;
	HTTPReply* reply;

	while (shouldContinue) {
		// is it time to send the next message?
		if (!shouldFinishUp && (!lastPost || (GetTimeAgeMS(lastPost) >= intervalMS))) {
			req = new HTTPRequest(reqTemplate);
			reply = manager->makeHTTPRequest(req, "localhost", port, encryption, 5000);
			delete(req);
			if (!reply) {
				LogPrint(0, LOG_SYSTEM, 0, "WebClient %u got no reply to HTTP request", id);
				serverError++;
			}
			else {
				//LogPrint(0, LOG_SYSTEM, 0, "WebClient %u got a reply to HTTP request: %s", id, HTTP_Status[reply->type]);
				switch (reply->type) {
				case HTTP_OK:
				case HTTP_USE_LOCAL_COPY:
					success++;
					break;
				case HTTP_MOVED_PERMANENTLY:
				case HTTP_MALFORMED_URL:
				case HTTP_UNAUTHORIZED:
				case HTTP_ACCESS_DENIED:
				case HTTP_FILE_NOT_FOUND:
				case HTTP_INTERNAL_ERROR:
				case HTTP_NOT_IMPLEMENTED:
				case HTTP_SERVICE_NOT_AVAILABLE:
					errors++;
					break;
				case HTTP_GATEWAY_TIMEOUT:
					timeout++;
					break;
				case HTTP_SERVER_UNAVAILABLE:
				case HTTP_SERVER_NOREPLY:
				case HTTP_SERVER_MALFORMED_REPLY:
				default:
					serverError++;
					break;
				}
			}
			delete(reply);
			lastPost = GetTimeNow();
			total++;
		}

		if (shouldFinishUp) {
			LogPrint(0, LOG_SYSTEM, 0, "[%u] finished total: %u  success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
				id, total, success, tooBusy, serverError, failed, timeout, errors);
			break;
		}

		if (GetTimeAgeMS(lastPrint) > printIntervalMS) {
			if (shouldFinishUp)
				LogPrint(0, LOG_SYSTEM, 0, "Client %u finishing up... (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
					id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
			else
				LogPrint(0, LOG_SYSTEM, 0, "[%u] (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
					id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
			lastPrint = GetTimeNow();
		}

		utils::Sleep(20);
	}

	delete manager;
	isRunning = false;
	return true;
}

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













TestWebSocketRequestClient::TestWebSocketRequestClient(uint32 id) {
	this->id = id;
	port = 10000;
	gatewayCount = 1;
	sendfreq = 1;
	reconnectMS = 0;
	payload = 1024;
	shouldContinue = true;
	isRunning = true;
	shouldFinishUp = false;
}

bool TestWebSocketRequestClient::init(uint16 port, uint32 gatewayCount, double sendfreq, int32 reconnectMS, uint32 payload, const char* longTestName) {
	uint32 myThreadID;
	this->port = port;
	if (gatewayCount)
		this->gatewayCount = gatewayCount;
	else
		this->gatewayCount = 1;
	this->sendfreq = sendfreq;
	this->reconnectMS = reconnectMS;
	this->payload = payload;
	this->longTestName = longTestName;
	return ThreadManager::CreateThread(TestWebSocketRequestClientRun, this, myThreadID, 0);
}

bool TestWebSocketRequestClient::finishUp() {
	LogPrint(0, LOG_SYSTEM, 0, "Client %u has been asked to finish up...", id);
	shouldFinishUp = true;
	return true;
}

bool TestWebSocketRequestClient::run() {

	int32 intervalMS = 1000;
	if (sendfreq > 0)
		intervalMS = (int32)(1000.0 / sendfreq);
	uint64 lastPost = 0;
	uint64 lastPrint = GetTimeNow();
	int32 printIntervalMS = 3000;

	uint32 stillWaiting, failed, timeout, errors, success, total, tooBusy, serverError;
	stillWaiting = failed = timeout = errors = success = total = tooBusy = serverError = 0;

	NetworkManager* manager = new NetworkManager();

	int8 encryption = NOENC;
	//payload = 40;
	char* bigdata = new char[payload];
	memset(bigdata, 1, payload);

	std::string uriString = utils::StringFormat("/longtest");

	WebsocketData* wsDataReply;
	JSONM* jmReply;
	WebsocketData* wsData = new WebsocketData();
	JSONM* jm = new JSONM();
//	jm->setJSON("{\"requestid\": 1}");
	jm->addData("Image1", bigdata, payload, "raw");
	//wsData->maskingKey = 2112143814;
	//wsData->setData(wsData->BINARY, true, jm->jmData, jm->getSize());
	//reqTemplate->createRequest(HTTP_POST, "", uriString.c_str(), bigdata, payload, false, 0);
	delete[] bigdata;
	//delete jm;

	NetworkChannel* channel = NULL;
	uint64 conID = 0;
	uint64 c = 0;

	while (shouldContinue) {
		// is it time to send the next message?
		if (!shouldFinishUp && (!lastPost || (GetTimeAgeMS(lastPost) >= intervalMS))) {

			std::string url = utils::StringFormat("http://localhost:%u/", port);

			if (!channel)
				channel = manager->createWebsocketConnection(url.c_str(), 0, NULL, conID, "JSONM");
			else if (!conID)
				conID = channel->createWebsocketConnection(url.c_str(), "JSONM");

			if (!conID) {
				LogPrint(0, 0, 1, "Couldn't connect to Websocket server on: %s", url.c_str());
				lastPost = GetTimeNow();
				continue;
			}

			jm->setJSON(utils::StringFormat("{\"requestid\": %llu, \"uri\": \"%s\", \"operation\": \"get\"}",
				++c, uriString.c_str()).c_str());
			wsData->setData(wsData->BINARY, true, jm->jmData, jm->getSize());

			//req = new HTTPRequest(reqTemplate);
			if (!channel->sendWebsocketData(wsData, conID)) {
				channel->endConnection(conID);
				LogPrint(0, 0, 1, "Lost connection to server...");
				lastPost = GetTimeNow();
				continue;
			}
			lastPost = GetTimeNow();
			total++;
		}

		if (conID) {
			if (wsDataReply = channel->waitForWebsocketData(conID, 20)) {
				jmReply = new JSONM(wsDataReply->data, wsDataReply->getPayloadSize());
				if (jmReply->isValid()) {
					success++;
				}
				else {
					failed++;
				}
				delete jmReply;
				delete wsDataReply;
			}
		}
		else
			utils::Sleep(20);

		if (shouldFinishUp) {
			LogPrint(0, LOG_SYSTEM, 0, "[%u] finished total: %u  success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
				id, total, success, tooBusy, serverError, failed, timeout, errors);
			break;
		}

		if (GetTimeAgeMS(lastPrint) > printIntervalMS) {
			if (shouldFinishUp)
				LogPrint(0, LOG_SYSTEM, 0, "Client %u finishing up... (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
					id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
			else
				LogPrint(0, LOG_SYSTEM, 0, "[%u] (%u/%u) success: %u  busy: %u  servererror: %u  failed: %u  timeout: %u  errors: %u\n",
					id, stillWaiting, total, success, tooBusy, serverError, failed, timeout, errors);
			lastPrint = GetTimeNow();
		}

		//utils::Sleep(20);
	}

	delete jm;
	delete wsData;
	delete manager;
	isRunning = false;
	return true;
}

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


/////////////////////////////////////////////////////////////
// Unit Test
/////////////////////////////////////////////////////////////
//
// Self-contained: stands up a local PROTOCOL_MESSAGE listener (the "gateway"
// server) on a high localhost port, points a RequestClient at it, exchanges a
// bounded set of request/reply messages, then shuts everything down. No
// external server, no infinite waits.
//
// The server side is driven from this same thread using the listener channel's
// polling API (waitForMessage / sendMessage): for every request the client
// posts, the server receives it, stamps it SUCCESS and echoes it straight back
// on the same connection. The client matches the reply to its outstanding
// request by reference, exactly as it would against a real gateway.

static bool Test_RequestClient() {
	const uint16 port = 38050;
	const uint32 reqCount = 20;
	const uint32 payload = 1024;

	unittest::progress(0, "starting server");

	NetworkManager* server = new NetworkManager();
	NetworkChannel* listen = server->createListener(port, NOENC, PROTOCOL_MESSAGE, true, 0, true, 0, NULL);
	if (!listen) {
		unittest::fail("could not start message listener on localhost:%u", port);
		delete server;
		return false;
	}

	unittest::progress(15, "connecting client");

	RequestClient* client = new RequestClient();
	if (!client->addGateway(1, "localhost", port, NOENC)) {
		unittest::fail("could not add gateway localhost:%u to client", port);
		delete client;
		delete server;
		return false;
	}

	// The client's addGateway sends a "ClientConnect" greeting on connect; the
	// server must drain it (it carries no request reference).
	uint64 recvid = 0;
	DataMessage* greeting = listen->waitForMessage(recvid, 3000);
	if (!greeting) {
		unittest::fail("server never received the client connect greeting");
		delete client;
		delete server;
		return false;
	}
	unittest::detail("received client greeting URI='%s' on conid %llu",
		greeting->getString("URI") ? greeting->getString("URI") : "", recvid);
	delete greeting;

	// The client only forwards requests once its connect event has landed
	// (lastConTime set); wait for that so the first postRequest is not dropped.
	if (!client->waitForConnection(3000)) {
		unittest::fail("client did not register as connected to localhost:%u", port);
		delete client;
		delete server;
		return false;
	}

	unittest::progress(40, "exchanging requests");

	char* dat = new char[payload];
	memset(dat, 7, payload);

	uint32 ok = 0;
	uint64 t0 = GetTimeNow();
	for (uint32 n = 0; n < reqCount; n++) {
		// Build and post a request; postRequest takes ownership of the message
		// and stamps it with a unique reference.
		DataMessage* req = new DataMessage();
		req->setString("URI", "unittest");
		req->setData("BigData", dat, payload);
		RequestReply* reply = client->postRequest(req);
		if (!reply) {
			unittest::fail("postRequest returned NULL on iteration %u", n);
			break;
		}

		// Server side: receive the forwarded request and echo it back as a
		// successful reply on the same connection (reference is preserved).
		uint64 conid = 0;
		DataMessage* got = listen->waitForMessage(conid, 3000);
		if (!got) {
			unittest::fail("server did not receive request %u", n);
			break;
		}
		got->setStatus(SUCCESS);
		if (!listen->sendMessage(got, conid)) {
			unittest::fail("server failed to send reply for request %u", n);
			delete got;
			break;
		}
		// sendMessage does not take ownership here.
		delete got;

		// Client side: wait for the matched reply.
		DataMessage* replyMsg = reply->waitForMessage(3000, true);
		if (!replyMsg) {
			unittest::fail("client got no reply for request %u", n);
			client->finishRequest(reply);
			break;
		}
		if (replyMsg->getStatus() != SUCCESS) {
			unittest::fail("reply %u had unexpected status %u", n, replyMsg->getStatus());
			delete replyMsg;
			client->finishRequest(reply);
			break;
		}
		delete replyMsg;
		client->finishRequest(reply);
		ok++;
		unittest::detail("request %u round-tripped ok", n);
		unittest::progress(40 + (n * 50) / reqCount, "exchanging requests");
	}
	double us = (double)(GetTimeNow() - t0);

	delete[] dat;

	unittest::progress(95, "shutting down");
	delete client;
	delete server;

	if (ok != reqCount)
		return false;

	if (us > 0)
		unittest::metric("request_throughput", (double)reqCount / us * 1e6, "req/s", true);
	unittest::metric("avg_request_latency", us / reqCount, "us", false);

	unittest::progress(100, "done");
	return true;
}

void Register_RequestClient_Tests() {
	UnitTestRunner::instance().registerTest(
		"requestclient", Test_RequestClient,
		"Request client/gateway round-trip over localhost", "network");
}

}
