

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

namespace cmlabs {

struct PsyType PsyAPI::CTRL_SYSTEM_READY =			{ { 1,10200,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PROCESS_INITIALISE =	{ { 1,10201,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PROCESS_GREETING =		{ { 1,10202,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PROCESS_SHUTDOWN =		{ { 1,10203,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_CONTEXT_CHANGE =		{ { 1,10297,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_SYSTEM_SHUTDOWN =		{ { 1,10298,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_SYSTEM_SHUTTINGDOWN =	{ { 1,10299,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_TRIGGER =				{ { 1,10300,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_QUERY =					{ { 1,10301,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_QUERY_REPLY =			{ { 1,10302,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_PULLCOMPONENTDATA =		{ { 1,10303,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_CREATECUSTOMPAGE =		{ { 1,10304,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_ADDSUBSCRIPTION =		{ { 1,10305,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_INTERSYSTEM_QUERY =		{ { 1,10998,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };
struct PsyType PsyAPI::CTRL_INTERSYSTEM_QUERY_REPLY = { { 1,10999,0,0,0,0,0,0,0,0,0,0,0,0,0,0 } };

PsyAPI::PsyAPI(PsySpace* space) {
	this->space = space;
	autoDelete = true;
	startedRunning = 0;
	shouldContinueRunning = false;
	currentCrankID = 0;
	lastCPUTicks = 0;
	currentCompID = 0;
	msgReceivedCount = 0;
	msgInputCount = 0;
	msgSentCount = 0;
	msgPostedCount = 0;
	commandlineBasedir[0] = 0;
	currentTriggerName[0] = 0;
	currentTriggerMsg = NULL;
	currentMsg = NULL;
	currentSignalMsg = NULL;
	currentTriggerSpec = NULL;
}

PsyAPI::~PsyAPI() {
	apiMutex.enter();
	shouldContinueRunning = false;
	this->space = NULL;
	delete(currentTriggerMsg);
	currentTriggerMsg = NULL;
	delete(currentMsg);
	currentMsg = NULL;
	delete(currentSignalMsg);
	currentSignalMsg = NULL;
	apiMutex.leave();
}

PsyAPI* PsyAPI::fromPython(unsigned long long ptr) {
	return (PsyAPI*)ptr;
}

const char* PsyAPI::fromPythonAddressOf(unsigned long long ptr) {
	return (const char*)ptr;
}

bool PsyAPI::isRunning() {
	apiMutex.enter();
	if (startedRunning) {
		apiMutex.leave();
		return true;
	}
	else {
		apiMutex.leave();
		return false;
	}
}

bool PsyAPI::shouldContinue() {
	apiMutex.enter();
	if (shouldContinueRunning) {
		apiMutex.leave();
		return true;
	}
	else {
		apiMutex.leave();
		return false;
	}
}

std::string PsyAPI::getCommandlineBasedir() {
	return commandlineBasedir;
}

bool PsyAPI::setCommandlineBasedir(const char* cmdlineBasedir) {
	utils::strcpyavail(commandlineBasedir, cmdlineBasedir, MAXCOMMANDLINELEN, true);
	return true;
}


bool PsyAPI::getModuleName(char* name, uint32 maxSize) {
	return space->manager->componentMemory->getComponentName(currentCompID, name, maxSize);
}

std::string PsyAPI::getModuleName() {
	return space->manager->componentMemory->getComponentNameString(currentCompID);
}

std::string PsyAPI::getOtherModuleName(uint32 id) {
	return space->manager->componentMemory->getComponentNameString(id);
}

bool PsyAPI::getCurrentScriptLanguage(char* language, uint32 maxSize) {
	return space->manager->dataMapsMemory->getCrankLanguage(currentCrankID, language, maxSize);
}

std::string PsyAPI::getCurrentScriptLanguage() {
	char* language = new char[1024];
	if (!space->manager->dataMapsMemory->getCrankLanguage(currentCrankID, language, 1024)) {
		delete[] language;
		return "";
	}
	std::string lang = language;
	delete[] language;
	return lang;
}

bool PsyAPI::getCurrentScript(char* script, uint32 maxSize) {
	return space->manager->dataMapsMemory->getCrankScript(currentCrankID, script, 1024);
}

std::string PsyAPI::getCurrentScript() {
	char* script = new char[MAXSCRIPTLEN];
	if (!space->manager->dataMapsMemory->getCrankScript(currentCrankID, script, MAXSCRIPTLEN)) {
		delete[] script;
		return "";
	}
	std::string scr = script;
	delete[] script;
	return scr;
}

bool PsyAPI::getCurrentScriptFilename(char* file, uint32 maxSize) {
	return space->manager->dataMapsMemory->getCrankLibraryFilename(currentCrankID, file, 1024);
}

std::string PsyAPI::getCurrentScriptFilename() {
	char* file = new char[1024];
	if (!space->manager->dataMapsMemory->getCrankLibraryFilename(currentCrankID, file, 1024)) {
		delete[] file;
		return "";
	}
	std::string filename = file;
	delete[] file;
	return filename;
}

bool PsyAPI::hasCurrentPostName(const char* name) {
	if (!apiMutex.enter(1000, __FUNCTION__))
		return false;
	if (currentTriggerSpec) {
		if (currentTriggerSpec->getPostSpec(name)) {
			apiMutex.leave();
			return true;
		}
	}
	apiMutex.leave();
	return false;
}

std::set<std::string> PsyAPI::getCurrentPostNames() {
	std::set<std::string> postNames;
	if (!apiMutex.enter(1000, __FUNCTION__))
		return postNames;
	if (currentTriggerSpec) {
		std::list<PostSpec*>* postSpecs = currentTriggerSpec->getPostSpecs(NULL);
		if (postSpecs && postSpecs->size()) {
			std::list<PostSpec*>::iterator it = postSpecs->begin(), itEnd = postSpecs->end();
			while (it != itEnd) {
				postNames.insert((*it)->name);
				it++;
			}
		}
		delete(postSpecs);
	}
	apiMutex.leave();
	return postNames;
}

uint32 PsyAPI::getInputQueueSize() {
	if (!apiMutex.enter(1000, __FUNCTION__))
		return 0;
	uint32 qSize = (uint32)inputQueue.size();
	apiMutex.leave();
//	printf("Pointer: %llu  Q: %u\n", (uint64)this, qSize);
	return qSize;
}

std::string PsyAPI::typeToText(PsyType type) {
	return space->manager->typeToText(type);
}

std::string PsyAPI::contextToText(PsyContext context) {
	return space->manager->contextToText(context);
}

bool PsyAPI::begin() {
	apiMutex.enter();

	shouldContinueRunning = true;
	startedRunning = GetTimeNow();
	utils::GetCPUTicks(lastCPUTicks);

	space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_STARTING,
		0, // cpu usage
		NULL, // no input message here, set in waitForNewMessage
		NULL, // no output message here
		1, // run count 1, only incremented at start
		0 // cycle count 0, only incremented at waitForNewMessage
		);
	apiMutex.leave();
	return true;
}

bool PsyAPI::finish() {

	apiMutex.enter();

	startedRunning = 0;
	currentCrankID = 0;
	currentCompID = 0;

	uint64 currentCPUTicks;
	utils::GetCPUTicks(currentCPUTicks);
	space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_FINISHED,
		currentCPUTicks - lastCPUTicks, // cpu usage
		NULL, // no input message here
		NULL, // no output message here
		0, // run count 0, only incremented at start
		0 // cycle count 0, only incremented at waitForMessage
		);
	//utils::GetCPUUsage(cpuUsageBefore);
	lastCPUTicks = currentCPUTicks;
	apiMutex.leave();
	return true;
}

bool PsyAPI::stop() {
	apiMutex.enter();
	shouldContinueRunning = false;
	apiMutex.leave();
	return true;
}



// ***************** PsyProbe plugins *****************
bool PsyAPI::addPsyProbeCustomView(const char* name, const char* templateURL) {
	return space->addPsyProbeCustomView(currentCompID, name, templateURL);
}




uint8 PsyAPI::addInputTrigger(uint32 crankID, uint32 compID, DataMessage* trigger, DataMessage* msg) {
	
//	printf("************ API Add Input Trigger %llu us old, Msg %llu us old...\n", GetTimeAge(trigger->getCreatedTime()), GetTimeAge(msg->getCreatedTime()));
	apiMutex.enter();

	msgReceivedCount++;
	if (startedRunning && currentCrankID) {
		if (crankID == currentCrankID) {
			// just add the trigger and msg and return
			inputQueue.push(TriggerAndMessage(trigger, msg));
			inputQueueSemaphore.signal();
			apiMutex.leave();
			return CRANKAPI_RUNNING;
		}
		else {
			apiMutex.leave();
			return CRANKAPI_INUSE;
		}
	}

	// reset measurements
	currentCrankID = crankID;
	currentCompID = compID;

	// add the trigger and msg
	inputQueue.push(TriggerAndMessage(trigger, msg));
	inputQueueSemaphore.signal();

	apiMutex.leave();
	return CRANKAPI_IDLE;
}

DataMessage* PsyAPI::waitForNewMessage(uint32 ms) {
	const char* triggerName;
	return waitForNewMessage(ms, triggerName);
}

std::string PsyAPI::getCurrentTriggerName() {
	return currentTriggerName;
}

PsyContext PsyAPI::getCurrentTriggerContext() {
	if (this->currentTriggerSpec)
		return this->currentTriggerSpec->context;
	else
		return NOCONTEXT;
}


DataMessage* PsyAPI::waitForNewMessage(uint32 ms, const char* &triggerName) {
	
	DataMessage* msg = NULL;
	uint32 s;
	uint64 currentCPUTicks;
	
//	if (!apiMutex.enter(ms ? ms : 10, __FUNCTION__))sig
	if (!apiMutex.enter(1000, __FUNCTION__))
		return NULL;
	
	if (!inputQueue.size()) {
		apiMutex.leave();
		// possible queue.size without mutex issue?
		if (!inputQueueSemaphore.wait(ms) || !inputQueue.size()) {
			utils::GetCPUTicks(currentCPUTicks);
			space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_RUNNING,
				currentCPUTicks - lastCPUTicks, // cpu usage
				NULL, // input message if any
				NULL, // no output message here
				0, // run count 0, only incremented at start
				1 // cycle count 
				);
			lastCPUTicks = currentCPUTicks;
			return NULL;
		}
//		if (!apiMutex.enter(ms, __FUNCTION__))
		if (!apiMutex.enter(1000, __FUNCTION__))
			return NULL;
	}

	delete(currentMsg);
	delete(currentTriggerMsg);

	TriggerAndMessage& tm = inputQueue.front();
	inputQueue.pop();

	while ((inputQueue.size() > 100) && (tm.second->getPolicy() & MESSAGE_NON_GUARANTEED)) {
		delete(tm.first);
		delete(tm.second);
		tm = inputQueue.front();
		inputQueue.pop();
	}

	currentTriggerMsg = tm.first;
	if (currentTriggerMsg) {
		currentTriggerSpec = (TriggerSpec*) currentTriggerMsg->getData("TriggerSpec", s);
		utils::strcpyavail(currentTriggerName, currentTriggerSpec->name, MAXKEYNAMELEN, true);
		triggerName = currentTriggerName;
	}
	else
		triggerName = NULL;
	msg = tm.second;
	if (autoDelete)
		currentMsg = msg;

//	printf("************ API Trigger %llu us old, Msg %llu us old...\n", GetTimeAge(currentTriggerMsg->getCreatedTime()), GetTimeAge(msg->getCreatedTime()));

	// calc stats
	apiMutex.leave();

	// Add stats to component
	utils::GetCPUTicks(currentCPUTicks);
	space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_RUNNING,
		currentCPUTicks - lastCPUTicks, // cpu usage
		msg, // input message if any
		NULL, // no output message here
		0, // run count 0, only incremented at start
		1 // cycle count 
		);
	lastCPUTicks = currentCPUTicks;

	msgInputCount++;
	return msg;
}

bool PsyAPI::setAutoDelete(bool autodel) {
	autoDelete = autodel;
	return true;
}

// Get current message receive time
uint64 PsyAPI::getCurrentMessageReceiveTime() {
	uint64 t;
	if (currentTriggerMsg && (t = currentTriggerMsg->getRecvTime()) )
		return t;
	else if (currentMsg)
		return currentMsg->getRecvTime();
	else
		return 0;
}

int32 PsyAPI::postOutputMessage(PsyType msgType, DataMessage* msg) {
	if (msgType == NOTYPE)
		return postOutputMessage(NULL, msg);
	if (msg) {
		msg->setType(msgType);
		return postOutputMessage(NULL, msg);
	}
	DataMessage* newMsg = new DataMessage(msgType, 0);
	bool result = postOutputMessage("A non-existing postname", newMsg);
	delete(newMsg);
	return result;
}


int32 PsyAPI::postOutputMessage(const char* postName, DataMessage* msg) {

	uint64 currentCPUTicks;

	apiMutex.enter();

	uint32 to;
	int32 failed = 0;
	int32 success = 0;
	if (!currentTriggerSpec) {
		if (!msg) {
			apiMutex.leave();
			logPrint(0, "Posting none - out of context");
			return POST_OUTOFCONTEXT;
		}
		if ( (to = msg->getTo()) || msg->getType().isValid()) {
			if (space->postMessage(msg)) {
				//logPrint(3,"Posted direct message to component %u", to);
				success++;
			}
			else {
				logPrint(0,"Failed to post direct message to component %u", to);
				failed++;
			}
		}
		else {
			apiMutex.leave();
			logPrint(0,"Posting none - out of context");
			return POST_OUTOFCONTEXT;
		}
	}
	else {
		PostSpec* postSpec;
		std::list<PostSpec*>* postSpecs = currentTriggerSpec->getPostSpecs(postName);
		if (!postSpecs || !postSpecs->size()) {
			delete(postSpecs);
			postSpecs = NULL;
			if (msg)
				postSpecs = currentTriggerSpec->getPostSpecs(msg->getType());
			if (!postSpecs || !postSpecs->size()) {
				delete(postSpecs);
				apiMutex.leave();
				logPrint(0, "Posting nothing - no post specs match for post name '%s'", postName);
				//logPrint(0, "Posting nothing - no post specs match for post name '%s'\n%s",
				//	postName, currentTriggerSpec->toXML().c_str());
				return POST_NOSPEC;
			}
		}

		uint64 msgTTL = 0;
		if (!msg)
			msg = new DataMessage();
		else
			msgTTL = msg->getTTL();

		// uint32 msgTTL = msg->getTTL();

		std::list<PostSpec*>::iterator it = postSpecs->begin(), itEnd = postSpecs->end();
		while (it != itEnd) {
			if (postSpec = *it) {
				if (postSpec->contextchange.isValid()) {
					msg->setType(CTRL_CONTEXT_CHANGE);
					msg->setContextChange(postSpec->contextchange);
					// msg->setData("Context", &postSpec->contextchange, sizeof(PsyContext));
				}
				else {
					msg->setType(postSpec->type);
				}
				postSpec->addContentToMsg(msg);
				//if (strlen(postSpec->contentKey) && !msg->getString(postSpec->contentKey))
				//	msg->setString(postSpec->contentKey, postSpec->content);
				msg->setMultiple(currentCompID, postSpec->to, postSpec->tag, msgTTL ? msgTTL : postSpec->ttl, postSpec->policy, GetTimeNow());

				if (space->postMessage(msg)) {
					//logPrint(3,"Posted msg delivered to %u", postSpec->to);
					success++;
				}
				else {
					logPrint(0,"Failed to deliver posted msg to %u", postSpec->to);
					failed++;
				}
			}
			it++;
		}
		delete(postSpecs);
	}

	msgPostedCount++;
	msgSentCount += success;

	if (!success)
		logPrint(0,"No msgs were delivered at postOutputMessage", success);
	//else
	//	logPrint(3,"Posted msg was delivered to %u receivers", success);

	// Add stats to component
	utils::GetCPUTicks(currentCPUTicks);
	space->manager->componentMemory->addComponentStats(currentCompID, COMPSTATUS_RUNNING,
		currentCPUTicks - lastCPUTicks, // cpu usage
		NULL, // no input message here
		msg, // output message
		0, // run count 0, only incremented at start
		0 // cycle count 0, only incremented at waitForNewMessage
		);
	// utils::GetCPUUsage(cpuUserBefore, cpuKernelBefore);
	lastCPUTicks = currentCPUTicks;

	apiMutex.leave();
	delete(msg);
	if (!success && failed)
		return POST_FAILED;
	else
		return success;
}

// ***************** Signals *****************
bool PsyAPI::emitSignal(const char* name, DataMessage* msg) {

	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		delete(msg);
		return false;
	}

	SignalSpec* signalSpec = currentTriggerSpec->getSignalSpec(name);
	if (!signalSpec) {
		apiMutex.leave();
		delete(msg);
		return false;
	}
	
	if (!msg)
		msg = new DataMessage();
	msg->setType(signalSpec->type);
	if (space->emitSignal(signalSpec->type, msg)) {
		apiMutex.leave();
		return true;
	}
	else {
		apiMutex.leave();
		delete(msg);
		return false;
	}
}

DataMessage* PsyAPI::waitForSignal(const char* name, uint32 timeout, uint64 lastReceivedTime) {

	apiMutex.enter(3000, __FUNCTION__);

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return NULL;
	}

	SignalSpec* signalSpec = currentTriggerSpec->getSignalSpec(name);
	if (!signalSpec) {
		apiMutex.leave();
		return NULL;
	}
	
	apiMutex.leave();
	DataMessage* msg = space->waitForSignal(signalSpec->type, timeout, lastReceivedTime);

	if (msg) {
		apiMutex.enter(3000, __FUNCTION__);
		if (currentSignalMsg)
			delete(currentSignalMsg);
		currentSignalMsg = msg;
		apiMutex.leave();
		return msg;
	}
	else
		return NULL;
}



uint8 PsyAPI::retrieve(std::list<DataMessage*> &result, const char* name, uint32 maxcount, uint32 maxage, uint32 timeout) {
	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
	if (!retrieveSpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	RetrieveSpec* specCopy = new RetrieveSpec;
	memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));

	if (maxcount)
		specCopy->maxcount = maxcount;
	if (maxage)
		specCopy->maxage = maxage;

	uint8 res = retrieve(result, specCopy, timeout);
	apiMutex.leave();
	return res;
}

uint8 PsyAPI::retrieveTimeParam(std::list<DataMessage*> &result, const char* name, uint64 startTime, uint64 endTime, uint32 maxcount, uint32 maxage, uint32 timeout) {
	std::list<DataMessage*> list;
	if (!startTime)
		return QUERY_FAILED;
	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
	if (!retrieveSpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	RetrieveSpec* specCopy = new RetrieveSpec;
	memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));

	specCopy->startTime = startTime;
	if (endTime)
		specCopy->endTime = endTime;
	if (maxcount)
		specCopy->maxcount = maxcount;
	if (maxage)
		specCopy->maxage = maxage;

	uint8 res = retrieve(result, specCopy, timeout);
	apiMutex.leave();
	return res;
}

uint8 PsyAPI::retrieveStringParam(std::list<DataMessage*> &result, const char* name, const char* startString, const char* endString, uint32 maxcount, uint32 maxage, uint32 timeout) {
	std::list<DataMessage*> list;
	if (!startString || !strlen(startString))
		return QUERY_FAILED;

	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
	if (!retrieveSpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	RetrieveSpec* specCopy = new RetrieveSpec;
	memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));

	utils::strcpyavail(specCopy->startString, startString, MAXVALUENAMELEN, true);
	if (endString)
		utils::strcpyavail(specCopy->endString, endString, MAXVALUENAMELEN, true);
	if (maxcount)
		specCopy->maxcount = maxcount;
	if (maxage)
		specCopy->maxage = maxage;

	uint8 res = retrieve(result, specCopy, timeout);
	apiMutex.leave();
	return res;
}

uint8 PsyAPI::retrieveIntegerParam(std::list<DataMessage*> &result, const char* name, int64 startInteger, int64 endInteger, uint32 maxcount, uint32 maxage, uint32 timeout) {
	std::list<DataMessage*> list;
	if (startInteger == INT64_NOVALUE)
		return QUERY_FAILED;

	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
	if (!retrieveSpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	RetrieveSpec* specCopy = new RetrieveSpec;
	memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));

	specCopy->startInt = startInteger;
	if (endInteger != INT64_NOVALUE)
		specCopy->endInt = endInteger;
	if (maxcount)
		specCopy->maxcount = maxcount;
	if (maxage)
		specCopy->maxage = maxage;

	uint8 res = retrieve(result, specCopy, timeout);
	apiMutex.leave();
	return res;
}

uint8 PsyAPI::retrieveFloatParam(std::list<DataMessage*> &result, const char* name, float64 startFloat, float64 endFloat, uint32 maxcount, uint32 maxage, uint32 timeout) {
	std::list<DataMessage*> list;
	if (startFloat == FLOAT64_NOVALUE)
		return QUERY_FAILED;
	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	RetrieveSpec* retrieveSpec = currentTriggerSpec->getRetrieveSpec(name);
	if (!retrieveSpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	RetrieveSpec* specCopy = new RetrieveSpec;
	memcpy(specCopy, retrieveSpec, sizeof(RetrieveSpec));

	specCopy->startFloat = startFloat;
	if (endFloat != FLOAT64_NOVALUE)
		specCopy->endFloat = endFloat;
	if (maxcount)
		specCopy->maxcount = maxcount;
	if (maxage)
		specCopy->maxage = maxage;

	uint8 res = retrieve(result, specCopy, timeout);
	apiMutex.leave();
	return res;
}

uint8 PsyAPI::retrieve(std::list<DataMessage*> &result, RetrieveSpec* spec, uint32 timeout) {
	if (!spec)
		return QUERY_FAILED;
	spec->origin = currentCompID;

	DataMessage* msg = new DataMessage(PsyAPI::CTRL_QUERY, this->currentCompID, spec->source, timeout);
	msg->setData("RetrieveSpec", (char*)spec, sizeof(RetrieveSpec));

	DataMessage* resultMsg = NULL;
	uint8 status = space->query(msg, &resultMsg, timeout);
	delete(msg);

	if (!resultMsg) {
		if (status == QUERY_FAILED)
			logPrint(0,"Query failed and did not return a message...");
		return status;
	}

	uint32 size;
	int64 count;
	const char* data = resultMsg->getData("ReplyData", size);
	if (!data || !resultMsg->getInt("ReplyCount", count)) {
		if (!data)
			logPrint(0,"Query did not return any ReplyData...");
		else
			logPrint(0,"Query did not return any ReplyCount...");
		delete(resultMsg);
		return QUERY_FAILED;
	}

	DataMessage* subMsg;
	const char* src = data;
	for (uint32 n=0; n<(uint32)count; n++) {
		if (GetObjID(src) != DATAMESSAGEID)
			break;
		subMsg = new DataMessage(src);
		result.push_back(subMsg);
		src += subMsg->getSize();
	}

	delete(spec);
	delete(resultMsg);
	return QUERY_SUCCESS;
}

bool PsyAPI::addSubscription(const char* xml) {
	apiMutex.enter();
	DataMessage* msg = new DataMessage(PsyAPI::CTRL_ADDSUBSCRIPTION, this->currentCompID, 0, 5000);
	msg->setString("Subscription", xml);

	DataMessage* resultMsg = NULL;
	uint8 status = space->query(msg, &resultMsg, 5000);
	delete(msg);

	if (resultMsg)
		delete(resultMsg);

	apiMutex.leave();
	return (status == QUERY_SUCCESS);
}


uint8 PsyAPI::queryCatalog(DataMessage** resultMsg, const char* name, DataMessage* msg, uint32 timeout) {
	*resultMsg = NULL;
	if (!msg)
		return QUERY_FAILED;
	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	QuerySpec* querySpec = currentTriggerSpec->getQuerySpec(name);
	if (!querySpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	if (strlen(querySpec->operation))
		msg->setString("Operation", querySpec->operation);
	if (strlen(querySpec->type))
		msg->setString("Type", querySpec->type);
	if (strlen(querySpec->ext))
		msg->setString("Ext", querySpec->ext);
	if (strlen(querySpec->subdir))
		msg->setString("Subdir", querySpec->subdir);
	if (strlen(querySpec->key))
		msg->setString("Key", querySpec->key);
	if (strlen(querySpec->value))
		msg->setString("Value", querySpec->value);
	if (querySpec->binary)
		msg->setString("Binary", "yes");

	if (strlen(querySpec->hostName)) {
		// This is a query for another system
		if (!querySpec->ipHost) {
			if (!utils::LookupIPAddress(querySpec->hostName, querySpec->ipHost)) {
				apiMutex.leave();
				return QUERY_COMPONENT_UNKNOWN;
			}
		}
		if (!querySpec->ipPort)
			querySpec->ipPort = 10000;
		// now we have all the info we need
		msg->setType(PsyAPI::CTRL_INTERSYSTEM_QUERY);
		msg->setFrom(currentCompID);
		msg->setTo(0);
		msg->setString("INTERSYSTEM_SOURCENAME", querySpec->sourceName);
		msg->setInt("INTERSYSTEM_ADDRESS", querySpec->ipHost);
		msg->setInt("INTERSYSTEM_PORT", querySpec->ipPort);
	}
	else {
		msg->setType(PsyAPI::CTRL_QUERY);
		msg->setFrom(currentCompID);
		msg->setTo(querySpec->source);
	}

	uint8 status = space->query(msg, resultMsg, timeout);
	delete(msg);

	apiMutex.leave();
	return status;
}


uint8 PsyAPI::queryCatalog(char** result, uint32 &resultsize, const char* name, const char* query, const char* operation, const char* data, uint32 datasize, uint32 timeout) {
	*result = NULL;
	apiMutex.enter();

	if (!currentTriggerSpec) {
		apiMutex.leave();
		return QUERY_FAILED;
	}

	QuerySpec* querySpec = currentTriggerSpec->getQuerySpec(name);
	if (!querySpec) {
		apiMutex.leave();
		return QUERY_NAME_UNKNOWN;
	}

	DataMessage* msg = new DataMessage();

	msg->setString("Name", querySpec->name);

	if (query && strlen(query))
		msg->setString("Query", query);
	if (data && datasize)
		msg->setData("Data", data, datasize);
	if (operation && strlen(operation))
		msg->setString("Operation", operation);
	else if (strlen(querySpec->operation))
		msg->setString("Operation", querySpec->operation);

	if (strlen(querySpec->type))
		msg->setString("Type", querySpec->type);
	if (strlen(querySpec->ext))
		msg->setString("Ext", querySpec->ext);
	if (strlen(querySpec->subdir))
		msg->setString("Subdir", querySpec->subdir);
	if (strlen(querySpec->key))
		msg->setString("Key", querySpec->key);
	if (strlen(querySpec->value))
		msg->setString("Value", querySpec->value);
	if (querySpec->binary)
		msg->setString("Binary", "yes");

	if (strlen(querySpec->hostName)) {
		// This is a query for another system
		if (!querySpec->ipHost) {
			if (!utils::LookupIPAddress(querySpec->hostName, querySpec->ipHost)) {
				apiMutex.leave();
				return QUERY_COMPONENT_UNKNOWN;
			}
		}
		if (!querySpec->ipPort)
			querySpec->ipPort = 10000;
		// now we have all the info we need
		msg->setType(PsyAPI::CTRL_INTERSYSTEM_QUERY);
		msg->setFrom(currentCompID);
		msg->setTo(0);
		msg->setString("INTERSYSTEM_SOURCENAME", querySpec->sourceName);
		msg->setInt("INTERSYSTEM_ADDRESS", querySpec->ipHost);
		msg->setInt("INTERSYSTEM_PORT", querySpec->ipPort);
	}
	else {
		msg->setType(PsyAPI::CTRL_QUERY);
		msg->setFrom(currentCompID);
		msg->setTo(querySpec->source);
	}

	DataMessage* resultMsg = NULL;
	uint8 status = space->query(msg, &resultMsg, timeout);
	delete(msg);

	if (resultMsg) {
		*result = resultMsg->getDataCopy("ReplyData", resultsize);
		delete(resultMsg);
	}

	apiMutex.leave();
	return status;
}

bool PsyAPI::queryReply(uint32 id, uint8 status, char* data, uint32 size, uint32 count) {
	DataMessage* msg = NULL;
	if (data && size) {
		msg = new DataMessage(PsyAPI::CTRL_QUERY_REPLY, currentCompID);
		msg->setData("ReplyData", data, size);
		msg->setInt("ReplyCount", count);
		// decide on EOL ###################
	}
	bool res = queryReply(id, status, msg);
	//delete(msg);
	return res;
}

bool PsyAPI::queryReply(uint32 id, uint8 status, DataMessage* msg) {
	apiMutex.enter();
	bool res = space->queryReply(id, status, msg);
	apiMutex.leave();
	return res;
}

bool PsyAPI::logPrint(uint8 level, const char *formatstring, ... ) {
	char* str;
	uint32 lenStr;
	va_list args;
	va_start(args, formatstring);
	str = utils::StringFormatVA(lenStr, formatstring, args);
	va_end(args);
	if (!str || !lenStr) {
		free(str);
		return false;
	}

	char* name = new char[256];
	if (!space->manager->componentMemory->getComponentName(currentCompID, name, 255)) {
		delete [] name;
		free(str);
		return false;
	}

	uint32 lenNewStr;
	char* newStr = utils::StringFormat(lenNewStr, "[%s] %s", name, str);
	delete [] name;
	free(str);

	LogEntry* entry = (LogEntry*) malloc(sizeof(LogEntry)+lenNewStr+1);
	entry->size = sizeof(LogEntry)+lenNewStr+1;
	entry->cid = LOGENTRYID;
	entry->time = GetTimeNow();
	entry->source = currentCompID;
	entry->subject = LOG_COMPONENT;
	entry->level = level;
	entry->type = LOGPRINT;
	entry->setText(newStr, lenNewStr);
	free(newStr);

	return space->logEntry(entry);
//	return LogPrint(this->currentCompID, LOG_COMPONENT, level, str);
}

} // namespace cmlabs
