

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

namespace	cmlabs{

/////////////////////////////////////////////////////////////
// Request Executor
/////////////////////////////////////////////////////////////

RequestExecutor::RequestExecutor() {
	lastRefID = 0;
	sentCount = 0;
	repliedCount = 0;
	shortReceivedCount = 0;
	longReceivedCount = 0;
	heartbeatIntervalMS = 3000;
	lastHeartbeat = 0;
	shouldContinue = true;
	isRunning = false;
	channel = NULL;
	manager = new NetworkManager();
	longReqLimit = 0;

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

RequestExecutor::~RequestExecutor() {
	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 RequestExecutor::addLongRequestName(const char* name) {
	if (channel)
		return false;
	longReqNames.push_back(name);
	return true;
}

bool RequestExecutor::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;

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

	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 (!con.conID) {
		mutex.leave();
		return false;
	}

	// Send connect message
	DataMessage* msg = new DataMessage();
	msg->setString("URI", "ExecutorConnect");
	msg->setString("LongRequests", utils::TextVectorConcat(longReqNames, ",", false).c_str());
	if (!channel->sendMessage(msg, con.conID)) {
		delete(msg);
		mutex.leave();
		return false;
	}
	
	delete(msg);
	gateways[con.conID] = con;
	mutex.leave();
	return true;
}


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

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

bool RequestExecutor::receiveMessage(DataMessage* msg, NetworkChannel* channel, uint64 conid) {
	if (!msg) return false;
	RequestReply* reply;
	RequestGatewayConnection con;
	bool isLongReq;
	std::map<uint64,RequestGatewayConnection>::iterator i;
	uint64 now = GetTimeNow();
	const char* req = msg->getString("URI");
	
	if (!mutex.enter(1000)) {
		// #########
		delete(msg);
		return true;
	}
	if ( (i=gateways.find(conid)) != gateways.end()) {
		reply = new RequestReply();
		reply->origin = conid;
		reply->execRef = ++lastRefID;
		reply->gatewayRef = msg->getReference();
		msg->setReference(reply->execRef);
		reply->giveRequestMessage(msg);
		reply->setStatus(QUEUED);
		requestMap[reply->execRef] = reply;

		if (longReqLimit)
			isLongReq = (msg->getSize() >= longReqLimit);
		else
			isLongReq = false;
		if (!isLongReq) {
			std::vector<std::string>::iterator il = longReqNames.begin(), el = longReqNames.end();
			while (il != el) {
				if (utils::stristr(req, il->c_str())) {
					isLongReq = true;
					break;
				}
				il++;
			}
		}
		//isLongReq = ( std::find(longReqNames.begin(), longReqNames.end(), req) != longReqNames.end());
		if (isLongReq) {
			longExecQ.add(msg);
			//longExecQSemaphore.signal();
			longReceivedCount++;
		}
		else {
			shortExecQ.add(msg);
			//shortExecQSemaphore.signal();
			shortReceivedCount++;
		}
		sendStatusNow();
	}
	else {
		// #############
		delete(msg);
	}
	mutex.leave();
	return true;
}

DataMessage* RequestExecutor::waitForLongRequest(uint32 timeoutMS) {
	DataMessage* msg;
	if (msg = longExecQ.waitForAndTakeFirst(timeoutMS)) {
		msg->setSendTime(GetTimeNow());
		return msg;
	}
	else
		return NULL;
}

DataMessage* RequestExecutor::waitForShortRequest(uint32 timeoutMS) {
	DataMessage* msg;
	if (msg = shortExecQ.waitForAndTakeFirst(timeoutMS)) {
		msg->setSendTime(GetTimeNow());
		return msg;
	}
	else
		return NULL;
}

bool RequestExecutor::replyToQuery(DataMessage* msg) {
	if (!msg) return false;
	uint64 execRef = msg->getReference();

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

	std::map<uint64,RequestReply*>::iterator i = requestMap.find(execRef);
	std::map<uint64,RequestGatewayConnection>::iterator con;

	if (i == requestMap.end()) {
		mutex.leave();
		return false;
	}

	if ( (con=gateways.find(i->second->origin)) == gateways.end()) {
		mutex.leave();
		return false;
	}

	replyQ.add(msg);
	repliedCount++;
	mutex.leave();
	return true;
}


bool RequestExecutor::replyToGateway(DataMessage* msg) {
	if (!msg)
		return false;
	uint64 execRef = msg->getReference();

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

	std::map<uint64,RequestReply*>::iterator i = requestMap.find(execRef);
	std::map<uint64,RequestGatewayConnection>::iterator con;

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

	if ( (con=gateways.find(reply->origin)) == gateways.end()) {
		// #############
		requestMap.erase(i);
		delete(reply);
		mutex.leave();
		//delete(msg);
		return false;
	}

	//printf("--- EX -> GW[%llu] ---\n", reply->origin);
	//fflush(stdout);

	msg->setReference(reply->gatewayRef);
	//DataMessage* msg2 = new DataMessage(*msg);
	//msg2->setReference(reply->gatewayRef);

	//msg->setReference(reply->gatewayRef);
	//DataMessage* msg2 = reply->peekRequestMessage();
	//msg2->setReference(reply->gatewayRef);

	if (channel->sendMessage(msg, reply->origin)) {
		// success
		//delete(msg);
		requestMap.erase(i);
		delete(reply);
		sentCount++;
		mutex.leave();
		return true;
	}
	else {
		requestMap.erase(i);
		delete(reply);
		mutex.leave();
		return false;
	}

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


bool RequestExecutor::run() {
	DataMessage* msg;
	isRunning = true;

	while (shouldContinue) {
		while (msg = replyQ.waitForAndTakeFirst(50)) {
			if (!mutex.enter(200, __FUNCTION__)) {
				delete(msg);
				break;
			}
			if (!replyToGateway(msg)) {
				// #######
			}
			delete(msg);
			mutex.leave();
		}

		// Send status heartbeat
		if (!lastHeartbeat || (GetTimeAgeMS(lastHeartbeat) > heartbeatIntervalMS) )
			sendStatusNow();
	}
	isRunning = false;
	return true;
}

bool RequestExecutor::sendStatusNow() {
	std::map<uint64,RequestGatewayConnection>::iterator i, e;
	DataMessage* msg = new DataMessage();
	msg->setString("URI", "ExecutorStatus");
	msg->setInt("ShortQSize", shortExecQ.getCount());
	msg->setInt("LongQSize", longExecQ.getCount());
	for (i=gateways.begin(), e=gateways.end(); i!=e; i++)
		channel->sendMessage(msg, i->second.conID);
	delete(msg);
	lastHeartbeat = GetTimeNow();
	return true;
}

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





TestRequestExecutor::TestRequestExecutor() : RequestExecutor() {
	uint32 myThreadID;
	ThreadManager::CreateThread(ShortExecutorRun, this, myThreadID, 0);
	ThreadManager::CreateThread(LongExecutorRun, this, myThreadID, 0);
}


bool TestRequestExecutor::shortRequestRun() {
	DataMessage* reply, *msg;
	while (shouldContinue) {
		if (msg = waitForShortRequest(100)) {
			//utils::Sleep(1);
			//reply = new DataMessage(*msg);
			reply = new DataMessage();
			//reply->setFrom(12);
			//reply->setString("Reques", "t");
			reply->setReference(msg->getReference());
			replyToQuery(reply);
			// do not delete msg, we don't own it
		}
	}
	return true;
}

bool TestRequestExecutor::longRequestRun() {
	DataMessage* reply, *msg;
	while (shouldContinue) {
		if (msg = waitForLongRequest(100)) {
			utils::Sleep(50);
			// reply = new DataMessage(*msg);
			reply = new DataMessage();
			//reply->setFrom(13);
			//reply->setString("REQUEST", "longtest");
			reply->setReference(msg->getReference());
			replyToQuery(reply);
			// do not delete msg, we don't own it
		}
	}
	return true;
}

THREAD_RET THREAD_FUNCTION_CALL ShortExecutorRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((TestRequestExecutor*)arg)->shortRequestRun() ? 0 : 1));
}

THREAD_RET THREAD_FUNCTION_CALL LongExecutorRun(THREAD_ARG arg) {
	if (arg == NULL) thread_ret_val(1);
	thread_ret_val((int)(((TestRequestExecutor*)arg)->longRequestRun() ? 0 : 1));
}

}
