

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  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 PsyAPI.cpp
	@brief Implementation of PsyAPI, the component-facing API handle (see PsyAPI.h).

	Also defines the well-known CTRL_* system control message types (PsyType
	constants in the 102xx/103xx/109xx range) used for system lifecycle, trigger
	delivery, queries and inter-system traffic.
*/
#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_TRIGGER_GROUP =			{ { 1,10307,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_RETRIEVESYSTEMIDS =		{ { 1,10306,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;
	lastWallTime = 0;
	chainCPUTicks = 0;
	chainWallTime = 0;
	chainCount = 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(2000);
	shouldContinueRunning = false;
	this->space = NULL;
	delete(currentTriggerMsg);
	currentTriggerMsg = NULL;
	delete(currentMsg);
	currentMsg = NULL;
	delete(currentSignalMsg);
	currentSignalMsg = NULL;
	// T1.6: free any owned grouped-set member messages.
	std::map<std::string, DataMessage*>::iterator git, gitEnd;
	for (git = currentGroupMap.begin(), gitEnd = currentGroupMap.end(); git != gitEnd; git++)
		if (git->second) delete(git->second);
	currentGroupMap.clear();
	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() {
	if (!apiMutex.enter(1000))
		return false;
	if (startedRunning) {
		apiMutex.leave();
		return true;
	}
	else {
		apiMutex.leave();
		return false;
	}
}

bool PsyAPI::shouldContinue() {
	if (!apiMutex.enter(1000))
		return true;
	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::hasCurrentRetrieveName(const char* name) {
	if (!apiMutex.enter(1000, __FUNCTION__))
		return false;
	if (currentTriggerSpec) {
		if (currentTriggerSpec->getRetrieveSpec(name)) {
			apiMutex.leave();
			return true;
		}
	}
	apiMutex.leave();
	return false;
}

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


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

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() {
	if (!apiMutex.enter(1000))
		return false;

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

	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() {

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

	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;
	lastWallTime = GetTimeNow();
	apiMutex.leave();
	return true;
}

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



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



uint64 PsyAPI::checkLastWaitForMessage() {
	if (!apiMutex.enter(1000))
		return CRANKAPI_FAILED;

	// If currently running
	if (startedRunning && currentCrankID) {
		if (inputQueue.size() && (GetTimeAgeMS(lastWallTime) > 5000)) {
			apiMutex.leave();
			return lastWallTime;
		}
	}
	apiMutex.leave();
	return 0;
}

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()));
	if (!apiMutex.enter(1000))
		return CRANKAPI_FAILED;

	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;
			lastWallTime = GetTimeNow();
			chainCPUTicks = 0;
			chainWallTime = 0;
			chainCount = 0;
			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();
	}

	DataMessage* msgFirst = tm.first;
	DataMessage* msgSecond = tm.second;

	// If this message is a trigger message (has tm.first) replace currentTriggerXXX
	if (msgFirst) {
		delete(currentTriggerMsg);
		currentTriggerMsg = msgFirst;
		currentTriggerSpec = (TriggerSpec*)currentTriggerMsg->getData("TriggerSpec", s);
		utils::strcpyavail(currentTriggerName, currentTriggerSpec->name, MAXKEYNAMELEN, true);
		triggerName = currentTriggerName;
	}
	else {
		// this is a temporary query message, keep the previous currentTriggerXXX around
		triggerName = NULL;
	}

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

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

	if (msgFirst && msgSecond) {
		chainCPUTicks = msgSecond->data->chaincputime;
		chainWallTime = msgSecond->data->chainwalltime;
		chainCount = msgSecond->data->chaincount;
	}
	else {
		chainCPUTicks = 0;
		chainWallTime = 0;
		chainCount = 0;
	}

	// 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;
	lastWallTime = GetTimeNow();

	msgInputCount++;
	return msg;
}


// T1.6: wait for a joined <triggergroup> set. Reuses waitForNewMessage() to dequeue
// (which sets currentTriggerMsg + currentTriggerName), then reads the member map that
// deliverTriggerGroup() attached under "Members". If the dequeued delivery is not a
// group activation, returns an empty map (and the group name is empty).
std::map<std::string, DataMessage*> PsyAPI::waitForNewMessageGroup(uint32 ms, std::string& groupName) {
	// Free any member messages we own from the previous grouped activation.
	std::map<std::string, DataMessage*>::iterator git, gitEnd;
	for (git = currentGroupMap.begin(), gitEnd = currentGroupMap.end(); git != gitEnd; git++)
		if (git->second) delete(git->second);
	currentGroupMap.clear();
	groupName.clear();

	const char* triggerName = NULL;
	DataMessage* primary = waitForNewMessage(ms, triggerName);
	// currentTriggerMsg is the delivery wrapper; it persists across timeout polls, so
	// only treat this as a group delivery when waitForNewMessage() actually dequeued a
	// fresh trigger this call (triggerName is non-NULL only for a fresh trigger msg).
	// Without this guard a completed group re-fires on every subsequent timeout poll.
	if (triggerName && currentTriggerMsg && currentTriggerMsg->getType() == CTRL_TRIGGER_GROUP) {
		currentGroupMap = currentTriggerMsg->getAttachedMessageMap("Members");
		groupName = triggerName;
	}
	else if (triggerName && primary) {
		// Not a group delivery: hand back the single message under its trigger name so a
		// crank calling only waitForNewMessageGroup() still sees something sensible.
		if (triggerName) groupName = triggerName;
		// Note: 'primary' is owned by the API (currentMsg); we do NOT copy it into the
		// owned map to avoid a double free. Callers mixing both APIs should prefer one.
	}
	return currentGroupMap;
}

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;
	utils::GetCPUTicks(currentCPUTicks);

	if (!apiMutex.enter(1000))
		return POST_FAILED;

	if (msg)
		msg->addTimeUsage((uint32)(currentCPUTicks - lastCPUTicks), (uint32)GetTimeAge(lastWallTime), chainCPUTicks, chainWallTime, chainCount + 1);

	uint32 currentTag = 0;
	if (currentMsg)
		currentTag = currentMsg->getTag();

	uint32 to;
	int32 failed = 0;
	int32 success = 0;
	if (!currentTriggerSpec) {
		if (!msg) {
			apiMutex.leave();
			logPrint(1, "Posting none - out of context");
			return POST_OUTOFCONTEXT;
		}
		if (currentTag && !msg->getTag())
			msg->setTag(currentTag);
		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(1,"Posting none - out of context");
			delete(msg);
			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(2, "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());
				delete(msg);
				return POST_NOSPEC;
			}
		}

		uint64 msgTTL = 0;
		if (!msg) {
			msg = new DataMessage();
			msg->addTimeUsage((uint32)(currentCPUTicks - lastCPUTicks), (uint32)GetTimeAge(lastWallTime), chainCPUTicks, chainWallTime, chainCount + 1);
		}
		else
			msgTTL = msg->getTTL();

		if (currentTag && !msg->getTag())
			msg->setTag(currentTag);

		// 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
	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) {

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

	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) {
	if (!apiMutex.enter(1000))
		return QUERY_FAILED;

	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;

	apiMutex.leave();
	uint8 res = retrieve(result, specCopy, timeout);
	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;
	if (!apiMutex.enter(1000))
		return QUERY_FAILED;

	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;

	apiMutex.leave();
	uint8 res = retrieve(result, specCopy, timeout);
	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;

	if (!apiMutex.enter(1000))
		return QUERY_FAILED;

	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;

	apiMutex.leave();
	uint8 res = retrieve(result, specCopy, timeout);
	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;

	if (!apiMutex.enter(1000))
		return QUERY_FAILED;

	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;

	apiMutex.leave();
	uint8 res = retrieve(result, specCopy, timeout);
	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;
	if (!apiMutex.enter(1000))
		return QUERY_FAILED;

	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;

	apiMutex.leave();
	uint8 res = retrieve(result, specCopy, timeout);
	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...");
		delete(spec);
		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);
		delete(spec);
		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) {
	if (!apiMutex.enter(1000))
		return false;
	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;
	if (!apiMutex.enter(1000))
		return QUERY_FAILED;

	if (!currentTriggerSpec) {
		logPrint(1, "Query '%s' cannot execute before a trigger message has been processed", name);
		apiMutex.leave();
		return QUERY_FAILED;
	}

	QuerySpec* querySpec = currentTriggerSpec->getQuerySpec(name);
	if (!querySpec) {
		logPrint(1, "Query '%s' not found in PsySpec", name);
		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);
	}

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

	return status;
}

uint8 PsyAPI::queryRemoteCatalog(DataMessage** resultMsg, const char* componentName, const char* ipAddress, uint16 port, DataMessage* msg, uint32 timeout) {
	uint32 addr = 0;
	if (!utils::LookupIPAddress(ipAddress, addr))
		return QUERY_NOT_AVAILABLE;
	return queryRemoteCatalog(resultMsg, componentName, addr, port, msg, timeout);
}

uint8 PsyAPI::queryRemoteCatalog(DataMessage** resultMsg, uint32& chosenAddress, const char* componentName, uint32* ipAddresses, uint32 numAddresses, uint16 port, DataMessage* msg, uint32 timeout) {
	if (chosenAddress > numAddresses)
		return QUERY_NOT_AVAILABLE;

	uint8 result;
	if (chosenAddress) {
		// try this address first
		result = queryRemoteCatalog(resultMsg, componentName, ipAddresses[chosenAddress-1], port, msg, timeout);
		if (result != QUERY_NOT_REACHABLE)
			return result;
		// if it failed, go back and try all again
	}

	chosenAddress = 0;
	for (uint32 n = 0; n < numAddresses; n++) {
		result = queryRemoteCatalog(resultMsg, componentName, ipAddresses[n], port, msg, timeout);
		if (result != QUERY_NOT_REACHABLE) {
			chosenAddress = n + 1;
			return result;
		}
		// else continue trying next address
	}
	// and give up if none of them connects
	return QUERY_NOT_REACHABLE;
}

uint8 PsyAPI::queryRemoteCatalog(DataMessage** resultMsg, const char* componentName, uint32 ipAddress, uint16 port, DataMessage* msg, uint32 timeout) {
	*resultMsg = NULL;
	if (!msg)
		return QUERY_FAILED;

	// This is a query for another system
	// we have all the info we need
	msg->setType(PsyAPI::CTRL_INTERSYSTEM_QUERY);
	msg->setFrom(currentCompID);
	msg->setTo(0);
	msg->setString("INTERSYSTEM_SOURCENAME", componentName);
	msg->setInt("INTERSYSTEM_ADDRESS", ipAddress);
	msg->setInt("INTERSYSTEM_PORT", port);

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

	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;
	if (!apiMutex.enter(1000, "Replying to query"))
		return QUERY_FAILED;

	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);
	}

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

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

	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) {
	if (!apiMutex.enter(1000, "Replying to query"))
		return false;
	bool res = space->queryReply(id, status, msg);
	apiMutex.leave();
	return res;
}

bool PsyAPI::logPrint(int 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 = (uint8)level;
	entry->type = LOGPRINT;
	entry->setText(newStr, lenNewStr);
	free(newStr);

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

} // namespace cmlabs
