

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

namespace	cmlabs{

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

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";
	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;
	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;
	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(200, __FUNCTION__))
		return LOCALERROR;

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

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

DataMessage* RequestReply::waitForMessage(uint32 timeoutMS, bool takeMessage) {
	DataMessage* msg;
	if (!mutex.enter(200, __FUNCTION__))
		return NULL;

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

	mutex.leave();
	if (semaphore.wait(timeoutMS)) {
		if (!mutex.enter(200, __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() {
	stop();

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

	delete(manager);
}

bool RequestClient::addGateway(uint32 id, std::string addr, uint16 port, uint8 encryption) {
	RequestGatewayConnection con;
	con.clear();
	con.id = id;
	con.addr = addr;
	con.port = port;
	con.encryption = encryption;
	connections.push_back(con);
	return true;
}

RequestReply* RequestClient::postRequest(DataMessage *msg) {
	if (!msg)
		return NULL;
	if (!mutex.enter(200, __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 true;
}

bool RequestClient::finishRequest(RequestReply* reply) {
	if (!reply || !reply->clientRef) return false;
	if (!mutex.enter(200, __FUNCTION__))
		return false;

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


bool RequestClient::receiveNetworkEvent(NetworkEvent* evt, NetworkChannel* channel, uint64 conid) {
	return false;
}

bool RequestClient::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
	if (!mutex.enter(200, __FUNCTION__))
		return false;
	uint64 ref = msg->getReference();
	std::map<uint64,RequestReply*>::iterator i = requestMap.find(ref);

	if (i == requestMap.end()) {
		// #############
		delete(msg);
		return true;
	}
	RequestReply* reply = i->second;
	reply->replyToRequest(msg, SUCCESS);
	receivedCount++;
	mutex.leave();
	return true;
}

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


bool RequestClient::run() {
	RequestGatewayConnection* con;
	RequestReply* reply;
	isRunning = true;

	while (shouldContinue) {
		//if (!mutex.enter(200, __FUNCTION__))
		//	continue;
		if (!channel && connections.size()) {
			con = &connections.front();
			channel = manager->createTCPConnection(con->addr.c_str(), con->port, con->encryption, PROTOCOL_MESSAGE, true, true, con->id, this, con->conID, con->location);
			if (channel) {
				// Send connect message
				DataMessage* msg = new DataMessage();
				msg->setString("URI", "ClientConnect");
				channel->sendMessage(msg, con->conID);
				delete(msg);
			}
		}

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

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

bool RequestClient::isConnected() {
	std::list<RequestGatewayConnection>::iterator i, e;
	for (i = connections.begin(), e = connections.end(); i != e; i++) {
		//printf("RUN0\n");fflush(stdout);
		if (channel->isConnected((*i).conID))
			return true;
	}
	return false;
}

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

	uint64 now = GetTimeNow();

	std::list<RequestGatewayConnection>::iterator i,e;
	for (i=connections.begin(), e=connections.end(); i!=e; i++) {
		//printf("RUN0\n");fflush(stdout);
		if (channel->isConnected((*i).conID)) {
			//printf("RUN1\n");fflush(stdout);
			if (channel->sendMessage(reply->peekRequestMessage(), (*i).conID)) {
				(*i).lastFailTime = 0;
				reply->setStatus(SENT);
				sentCount++;
				//printf("RUN2\n");fflush(stdout);
				return true;
			}
			else {
				(*i).lastFailTime = GetTimeNow();
				//printf("RUN3\n");fflush(stdout);
				continue;
			}
		}
		else if (!(*i).conID) {
			if ((*i).conID = channel->createTCPConnection((*i).addr.c_str(), (*i).port, (*i).encryption, PROTOCOL_MESSAGE, true, true, (*i).location)) {
				DataMessage* msg = new DataMessage();
				msg->setString("URI", "ClientConnect");
				channel->sendMessage(msg, (*i).conID);
				delete(msg);
				//printf("RUN10\n");fflush(stdout);
				if (channel->sendMessage(reply->peekRequestMessage(), (*i).conID)) {
					//printf("RUN11\n");fflush(stdout);
					reply->setStatus(SENT);
					sentCount++;
					(*i).lastConTime = GetTimeNow();
					(*i).lastFailTime = 0;
					return true;
				}
				//printf("RUN12\n");fflush(stdout);
			}
		}
	}
	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));
}











TestRequestClient::TestRequestClient(bool isLong) : RequestClient() {
	uint32 myThreadID;
	ThreadManager::CreateThread(TestRequestClientRun, this, myThreadID, 0);
}


bool TestRequestClient::testRun() {
	//DataMessage* reply, *msg;
	while (shouldContinue) {
		utils::Sleep(20);
	}
	return true;
}

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



}
