

//_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/_/
//_/_/
//_/_/  This file is part of PSYCLONE 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/
//_/_/
//_/_/  PSYCLONE 2.1 is released under the GNU Lesser General Public License 
//_/_/  (LGPL) version 3.0, and is subject to the following
//_/_/  
//_/_/  - 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. 
//_/_/    
//_/_/  A commercial license for PSYCLONE 2.1 is also available to 
//_/_/  any interested party from the copyright holder.
//_/_/  
//_/_/  You can redistribute this copy of PSYCLONE 2.1 and/or modify it under 
//_/_/  the terms of the GNU General Public License as published by the Free 
//_/_/  Software Foundation, either version 3 of the License, or (at your 
//_/_/  option) any later version.
//_/_/  
//_/_/  You should have received a copy of the GNU General Public License
//_/_/  and the GNU Lesser General Public License along with PSYCLONE 2.1.  
//_/_/  If not, see <http://www.gnu.org/licenses/>.
//_/_/  
//_/_/  PSYCLONE 2.1 is distributed in the hope that it will be useful,
//_/_/  but WITHOUT ANY WARRANTY; without even the implied warranty of
//_/_/  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
//_/_/  GNU General Public License for more details.
//_/_/
//_/_/  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 "SubscriptionEngine.h"
#include "Node.h"

namespace	cmlabs{

SubTriggers::SubTriggers(SubscriptionEngine* engine, uint8 level, uint16 subType) {
	this->engine = engine;
	this->level = level;
	this->subType = subType;
}

SubTriggers::~SubTriggers() {
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++)
		delete(it->second);
	//levelTriggers.clear();
	std::list<TriggerSpec*>::iterator tit, titEnd;
	for (tit = levelTriggers.begin(), titEnd = levelTriggers.end(); tit != titEnd; tit++)
		delete(*tit);
	engine = NULL;
}

// aaa.bbb.ccc.ddd
// aaa.bbb.ccc
// aaa.bbb
// aaa.***.ccc.ddd

std::string SubTriggers::printFriendlyTriggersForMsg(DataMessage* msg) {

	std::string str = utils::StringFormat("Checking msg type %s (%s) against level %u\n",
		engine->typeToText(msg->getType()).c_str(), msg->getType().toString().c_str(), level);

	// Insert all at this level
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
		if (*it)
			str += printTriggerFriendly(*it, "");
		//else
		//	str += utils::StringFormat("  NULL\n");
	}
	
	SubTriggers* sub;

	if (sub = subTriggers[msg->getType()[level]]) {
		str += utils::StringFormat("Found matching sub for level %u\n", level);
		str += sub->printFriendlyTriggersForMsg(msg);
	}
	else
		str += utils::StringFormat("Didn't find matching sub for level %u\n", level);

	if (sub = subTriggers[0xFFFF]) {
		str += utils::StringFormat("Found matching sub for wildcard level\n");
		str += sub->printFriendlyTriggersForMsg(msg);
	}

	return str;
}

uint32 SubTriggers::getTriggersForMsg(DataMessage* msg, std::list<TriggerSpec*>* triggers) {

	uint16 msgTypeLevelValue = msg->getType()[level];

	uint32 count = 0;
	if (!msgTypeLevelValue) {
		if (count = (uint32)levelTriggers.size())
			triggers->insert(triggers->end(), levelTriggers.begin(), levelTriggers.end());
	}
	
	SubTriggers* sub;
	std::map<uint16, SubTriggers*>::iterator si;
	// find() not operator[] on this read (routing) path: operator[] inserts a
	// NULL element on a miss, growing the map and mutating it concurrently with
	// other routing threads reading it under the engine's shared read lock.
	if ((si = subTriggers.find(msgTypeLevelValue)) != subTriggers.end() && (sub = si->second))
		count += sub->getTriggersForMsg(msg, triggers);
	if ((si = subTriggers.find(0xFFFF)) != subTriggers.end() && (sub = si->second))
		count += sub->getTriggersForMsg(msg, triggers);
	return count;
}

bool SubTriggers::addSubscription(uint32 compID, TriggerSpec* trigger) {

	TriggerSpec* copy;

	// Does the trigger go into this level
	if (trigger->type[level] == 0) {
		// Check if the same trigger already exists
		bool didExist = false;
		std::list<TriggerSpec*>::iterator i = levelTriggers.begin(), e = levelTriggers.end();
		while (i != e) {
			if (((*i)->componentID == trigger->componentID) &&
				((*i)->type == trigger->type) &&
				((*i)->context == trigger->context)) {
				delete (*i);
				levelTriggers.erase(i);
				didExist = true;
				break;
			}
			i++;
		}
		copy = (TriggerSpec*) malloc(trigger->size);
		memcpy(copy, trigger, trigger->size);
		levelTriggers.push_back(copy);
		if (!didExist)
			engine->registerSubTrigger(this, trigger->componentID);
		return true;
	}

	SubTriggers* sub;
	uint16 subType = trigger->type[level];
	if (!(sub = subTriggers[subType]))
		subTriggers[subType] = sub = new SubTriggers(engine, level+1, subType);
	return sub->addSubscription(compID, trigger);
}

bool SubTriggers::removeSubscription(uint32 compID) {
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd;) {
		if ((*it)->componentID == compID) {
			delete(*it);
			it = levelTriggers.erase(it);
		}
		else
			it++;
	}
	return true;
}

TriggerSpec* SubTriggers::findTrigger(uint32 compID, const char* name) {
	TriggerSpec* triggerSpec;
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
		if (*it && ((*it)->componentID == compID) && (stricmp((*it)->name, name) == 0))
			return *it;
	}
	std::map<uint16, SubTriggers*>::iterator sit, sitEnd;
	for (sit = subTriggers.begin(), sitEnd = subTriggers.end(); sit != sitEnd; sit++) {
		if (sit->second && (triggerSpec = sit->second->findTrigger(compID, name)))
			return triggerSpec;
	}
	return NULL;
}

std::list<PostSpec*>* SubTriggers::findPosts(uint32 compID, const char* name) {
	std::list<PostSpec*>* postSpecs;
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
		if (*it && ((*it)->componentID == compID) && (postSpecs = (*it)->getPostSpecs(name)))
			return postSpecs;
	}
	std::map<uint16, SubTriggers*>::iterator sit, sitEnd;
	for (sit = subTriggers.begin(), sitEnd = subTriggers.end(); sit != sitEnd; sit++) {
		if (sit->second && (postSpecs = sit->second->findPosts(compID, name)))
			return postSpecs;
	}
	return NULL;
}


bool SubTriggers::getAllTriggers(std::map<uint32, std::list<TriggerSpec*> > &triggerMap) {
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
		if (*it)
			triggerMap[(*it)->componentID].push_back(*it);
	}

	std::map<uint16, SubTriggers*>::iterator sit, sitEnd;
	for (sit = subTriggers.begin(), sitEnd = subTriggers.end(); sit != sitEnd; sit++) {
		if (sit->second) {
			if (!sit->second->getAllTriggers(triggerMap))
				return false;
		}
	}
	return true;
}

std::string SubTriggers::printFriendly() {
	char tab[64];
	memset(tab, 32, 2*level);
	tab[2*level] = 0;
	std::string str; // = utils::StringFormat("Level: %u Count: %u\n", level, levelTriggers.size());
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
		if (*it)
			str += printTriggerFriendly(*it, tab);
		//else
		//	str += utils::StringFormat("  NULL\n");
	}
	std::map<uint16, SubTriggers*>::iterator sit, sitEnd;
	for (sit = subTriggers.begin(), sitEnd = subTriggers.end(); sit != sitEnd; sit++) {
		if (sit->second)
			str += utils::StringFormat("%s%s (%u)\n%s", tab, engine->subTypeToText(sit->first).c_str(), sit->first, sit->second->printFriendly().c_str());
		//else
		//	str += utils::StringFormat("SubType[%u]: %s NULL\n", sit->first, engine->subTypeToText(sit->first).c_str());
	}
	return str;
}

std::string SubTriggers::printFriendlyHTML() {
	std::string str; // = utils::StringFormat("Level: %u Count: %u\n", level, levelTriggers.size());
	std::list<TriggerSpec*>::iterator it, itEnd;
	if (levelTriggers.size()) {
		for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
			if (*it)
				str += printTriggerFriendlyHTML(*it);
			//else
			//	str += utils::StringFormat("  NULL\n");
		}
	}
	if (subTriggers.size())
		str += "<table border=1>";
	std::map<uint16, SubTriggers*>::iterator sit, sitEnd;
	for (sit = subTriggers.begin(), sitEnd = subTriggers.end(); sit != sitEnd; sit++) {
		if (sit->second)
			str += utils::StringFormat("<tr><td>%s (%u)</td><td>%s</td></tr>", engine->subTypeToText(sit->first).c_str(), sit->first, sit->second->printFriendlyHTML().c_str());
	}
	if (subTriggers.size())
		str += "</table>\n";
	return str;
}

std::string SubTriggers::toXML() {
	if (!levelTriggers.size() && !subTriggers.size())
		return utils::StringFormat("<subtrigger subtype=\"%u\" level=\"%u\" leveltriggercount=\"0\" subtriggercount=\"0\" />\n", subType, level);

	uint32 levelTriggerCount = 0, subTriggerCount = 0;

	std::string str;
	std::list<TriggerSpec*>::iterator it, itEnd;
	if (levelTriggers.size()) {
		for (it = levelTriggers.begin(), itEnd = levelTriggers.end(); it != itEnd; it++) {
			if (*it) {
				str += (*it)->toXML();
				levelTriggerCount++;
			}
		}
	}
	std::map<uint16, SubTriggers*>::iterator sit, sitEnd;
	for (sit = subTriggers.begin(), sitEnd = subTriggers.end(); sit != sitEnd; sit++) {
		if (sit->second) {
			str += sit->second->toXML();
			subTriggerCount++;
		}
	}

	return utils::StringFormat("<subtrigger subtype=\"%u\" level=\"%u\" leveltriggercount=\"%u\" subtriggercount=\"%u\">\n%s</subtrigger>\n",
		subType, level, levelTriggerCount, subTriggerCount, str.c_str());
}


std::string SubTriggers::printTriggerFriendly(TriggerSpec* trigger, const char* tab) {

	uint32 n;
	std::string str = utils::StringFormat("%s%s  Comp: %u, %u filters, %u signals, %u posts, context: %s\n",
		tab,
		engine->typeToText(trigger->type).c_str(),
		trigger->componentID,
		trigger->filterCount,
		trigger->signalCount,
		trigger->postCount,
		engine->contextToText(trigger->triggerContext).c_str()
		);

	for (n=0; n<trigger->retrieveCount; n++) {
		RetrieveSpec* rSpec = trigger->getRetrieveSpec(n);
		if (rSpec)
			str += utils::StringFormat("%s Retrieve[%u] %s: type: %s, context: %s\n",
				tab, n,
				rSpec->name,
				engine->typeToText(rSpec->type).c_str(),
				engine->contextToText(rSpec->context).c_str()
				);
	}

	for (n=0; n<trigger->signalCount; n++) {
		SignalSpec* sSpec = trigger->getSignalSpec(n);
		if (sSpec)
			str += utils::StringFormat("%s Signal[%u] %s: type: %s\n",
				tab, n,
				sSpec->name,
				engine->typeToText(sSpec->type).c_str()
				);
	}

	for (n=0; n<trigger->queryCount; n++) {
		QuerySpec* qSpec = trigger->getQuerySpec(n);
		if (qSpec)
			str += utils::StringFormat("%s Query[%u] %s: type: %s, operation: %s\n",
				tab, n,
				qSpec->name,
				qSpec->type,
				qSpec->operation
				);
	}

	for (n=0; n<trigger->postCount; n++) {
		PostSpec* pSpec = trigger->getPostSpec(n);
		if (pSpec)
			str += utils::StringFormat("%s Post[%u] %s: type: %s, context: %s\n",
				tab, n,
				pSpec->name,
				engine->typeToText(pSpec->type).c_str(),
				engine->contextToText(pSpec->context).c_str()
				);
	}

	return str;
}


std::string SubTriggers::printTriggerFriendlyHTML(TriggerSpec* trigger) {

	uint32 n;
	std::string str = utils::StringFormat("<B>%s</B> (%u), %u filters, %u signals, %u posts, context: %s<ul>",
		// engine->typeToText(trigger->type).c_str(),
		engine->getComponentName(trigger->componentID).c_str(),
		trigger->componentID,
		trigger->filterCount,
		trigger->signalCount,
		trigger->postCount,
		engine->contextToText(trigger->triggerContext).c_str()
		);

	for (n=0; n<trigger->retrieveCount; n++) {
		RetrieveSpec* rSpec = trigger->getRetrieveSpec(n);
		if (rSpec)
			str += utils::StringFormat("<li>Retrieve[%u] %s: type: %s, context: %s</li>",
				n,
				rSpec->name,
				engine->typeToText(rSpec->type).c_str(),
				engine->contextToText(rSpec->context).c_str()
				);
	}

	for (n=0; n<trigger->signalCount; n++) {
		SignalSpec* sSpec = trigger->getSignalSpec(n);
		if (sSpec)
			str += utils::StringFormat("<li>Signal[%u] %s: type: %s</li>",
				n,
				sSpec->name,
				engine->typeToText(sSpec->type).c_str()
				);
	}

	for (n=0; n<trigger->queryCount; n++) {
		QuerySpec* qSpec = trigger->getQuerySpec(n);
		if (qSpec)
			str += utils::StringFormat("<li>Query[%u] %s: type: %s, operation: %s</li>",
				n,
				qSpec->name,
				qSpec->type,
				qSpec->operation
				);
	}

	for (n=0; n<trigger->postCount; n++) {
		PostSpec* pSpec = trigger->getPostSpec(n);
		if (pSpec)
			str += utils::StringFormat("<li>Post[%u] %s: type: %s, context: %s</li>",
				n,
				pSpec->name,
				engine->typeToText(pSpec->type).c_str(),
				engine->contextToText(pSpec->context).c_str()
				);
	}

	str += "</ul>\n";
	return str;
}



// *********************************************************************************


SubscriptionEngine::SubscriptionEngine(Node* node) {
	this->node = node;
	currentReaders = 0;
}

SubscriptionEngine::~SubscriptionEngine() {
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++)
		delete(it->second);
	//levelTriggers.clear();
	this->node = NULL;
}

std::string SubscriptionEngine::contextToText(PsyContext context) {
	return node->manager->contextToText(context);
}
std::string SubscriptionEngine::subContextToText(uint16 subcontext) {
	return node->manager->subContextToText(subcontext);
}

std::string SubscriptionEngine::typeToText(PsyType type) {
	return node->manager->typeToText(type);
}
std::string SubscriptionEngine::subTypeToText(uint16 subType) {
	return node->manager->subTypeToText(subType);
}
std::string SubscriptionEngine::getComponentName(uint32 compID) {
	return node->manager->getComponentName(compID);
}

bool SubscriptionEngine::syncSubscriptions(DataMessage* msg) {
	// Extract all triggers and add them to the subscription
	std::map<uint32, std::list<TriggerSpec*> > triggerMap;

	TriggerSpec* spec;
	uint32 n = 0, size;
	while (spec = (TriggerSpec*)msg->getData((int64)n, "Triggers", size)) {
		triggerMap[spec->componentID].push_back(spec);
		n++;
	}

	bool success = true;
	std::map<uint32, std::list<TriggerSpec*> >::iterator i = triggerMap.begin(), e = triggerMap.end();
	while (i != e) {
		if (success && !addSubscription(i->first, i->second))
			success = false;
		i++;
	}
	return success;
}

bool SubscriptionEngine::syncContexts(DataMessage* msg) {
	uint32 n = 0;
	const char* contextName;
	PsyContext context;
	while (contextName = msg->getString((int64)n, "ActiveContexts")) {
		if (context.fromString(contextName))
			setContextActive(context, NOCONTEXT);
		n++;
	}
	n = 0;
	while (contextName = msg->getString((int64)n, "InactiveContexts")) {
		if (context.fromString(contextName))
			setContextInactive(context);
		n++;
	}
	return true;
}

bool SubscriptionEngine::writeContextsToMsg(DataMessage* msg) {
	if (!activeContexts.size())
		return true;
	std::map<PsyContext, uint64>::iterator it, itEnd;
	uint32 count = 0;
	for (it = activeContexts.begin(), itEnd = activeContexts.end(); it != itEnd; it++) {
		if (((PsyContext)it->first).isValid()) {
			msg->setString(count, "ActiveContexts", ((PsyContext)it->first).toString().c_str());
			count++;
		}
	}
	count = 0;
	for (it = inactiveContexts.begin(), itEnd = inactiveContexts.end(); it != itEnd; it++) {
		if (((PsyContext)it->first).isValid()) {
			msg->setString(count, "InactiveContexts", ((PsyContext)it->first).toString().c_str());
			count++;
		}
	}
	return true;
}

bool SubscriptionEngine::writeTriggersToMsg(DataMessage* msg) {
	// Get list of all triggers
	// wait for any writes to finish;
	writeMutex.enter();
	entranceMutex.enter();
	currentReaders++;
	entranceMutex.leave();
	writeMutex.leave();

	bool success = true;

	std::map<uint32, std::list<TriggerSpec*> > triggerMap;
	if (!getAllTriggers(triggerMap))
		success = false;

	// Write all triggers to msg
	uint32 n = 0;
	std::map<uint32, std::list<TriggerSpec*> >::iterator i = triggerMap.begin(), e = triggerMap.end();
	std::list<TriggerSpec*>::iterator ti, te;
	while (i != e) {
		ti = i->second.begin();
		te = i->second.end();
		while (ti != te) {
			msg->setData(n, "Triggers", (char*)(*ti), (*ti)->size);
			ti++;
			n++;
		}
		i++;
	}
	
	// release read access
	entranceMutex.enter();
	currentReaders--;
	entranceMutex.leave();

	return success;
}


bool SubscriptionEngine::addSubscription(uint32 compID, std::list<TriggerSpec*> &triggers) {

	// wait for write access
	writeMutex.enter();
	entranceMutex.enter();
	while (true) {
		if (currentReaders) {
			entranceMutex.leave();
			utils::Sleep(20);
			entranceMutex.enter();
		}
		else {
			entranceMutex.leave();
			break;
		}
	}

	if (compIDSubTriggers[compID].size()) {
		// Find all subtriggers registered with compID
		std::list<SubTriggers*>::iterator sit, sitEnd;
		for (sit = compIDSubTriggers[compID].begin(), sitEnd = compIDSubTriggers[compID].end(); sit != sitEnd; sit++) {
			// Remove triggers from their levelTriggers with this compID
			(*sit)->removeSubscription(compID);
		}
		compIDSubTriggers[compID].clear();
	}

	bool result = true;
	std::list<TriggerSpec*>::iterator it, itEnd;
	for (it = triggers.begin(), itEnd = triggers.end(); it != itEnd; it++) {
		if (!addSubscription(compID, *it))
			result = false;;
	}

	// release write access
	writeMutex.leave();

	return result;
}

bool SubscriptionEngine::removeSubscription(uint32 compID) {

	// wait for write access
	writeMutex.enter();
	entranceMutex.enter();
	while (true) {
		if (currentReaders) {
			entranceMutex.leave();
			utils::Sleep(20);
			entranceMutex.enter();
		}
		else {
			entranceMutex.leave();
			break;
		}
	}

	// Find all subtriggers registered with compID
	std::list<SubTriggers*>::iterator sit, sitEnd;
	for (sit = compIDSubTriggers[compID].begin(), sitEnd = compIDSubTriggers[compID].end(); sit != sitEnd; sit++) {
		// Remove triggers from their levelTriggers with this compID
		(*sit)->removeSubscription(compID);
	}
	compIDSubTriggers[compID].clear();

	// release write access
	writeMutex.leave();

	return true;
}

bool SubscriptionEngine::registerSubTrigger(SubTriggers* sub, uint32 compID) {
	// check if already exists
	std::list<SubTriggers*>::iterator i = compIDSubTriggers[compID].begin(), e = compIDSubTriggers[compID].end();
	while (i != e) {
		if ((*i) == sub)
			return true;
		i++;
	}
	// record subtrigger for compID
	compIDSubTriggers[compID].push_back(sub);
	return true;
}

bool SubscriptionEngine::addSubscription(uint32 compID, TriggerSpec* trigger) {
	if (!trigger->type.isValid())
		return false;

	SubTriggers* sub;
	uint16 subType = trigger->type[0];
	if (!(sub = subTriggers[subType]))
		subTriggers[subType] = sub = new SubTriggers(this, 1, subType);
	return sub->addSubscription(compID, trigger);
}

bool SubscriptionEngine::getAllTriggers(std::map<uint32, std::list<TriggerSpec*> > &triggerMap) {
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it != itEnd; it++) {
		if (it->second) {
			if (!it->second->getAllTriggers(triggerMap))
				return false;
		}
	}
	return true;
}

std::string SubscriptionEngine::printFriendly() {
	std::string str;
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++) {
		if (it->second)
			str += utils::StringFormat("%s (%u)\n%s", subTypeToText(it->first).c_str(), it->first, it->second->printFriendly().c_str());
		//else
		//	str += utils::StringFormat("SubType[%u/top]: %s NULL\n", it->first, subTypeToText(it->first).c_str());
	}

	std::map<PsyContext, uint64>::iterator i, e;
	uint32 count = 0;
	for (i = activeContexts.begin(), e = activeContexts.end(); i != e; i++) {
		if (((PsyContext)i->first).isValid()) {
			str += utils::StringFormat("Active context: %s (%s)\n",
				((PsyContext)i->first).toString().c_str(),
				PrintTimeString(i->second).c_str());
			count++;
		}
	}
	count = 0;
	for (i = inactiveContexts.begin(), e = inactiveContexts.end(); i != e; i++) {
		if (((PsyContext)i->first).isValid()) {
			str += utils::StringFormat("Inactive context: %s (%s)\n",
				((PsyContext)i->first).toString().c_str(),
				PrintTimeString(i->second).c_str());
			count++;
		}
	}

	return str;
}

std::string SubscriptionEngine::printFriendlyHTML() {
	std::string str;
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	str += "<table border=1>";
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++) {
		if (it->second) {
			str += utils::StringFormat("<tr><td>%s (%u)</td><td>%s</td></tr>", subTypeToText(it->first).c_str(), it->first, it->second->printFriendlyHTML().c_str());
		}
		//else
		//	str += utils::StringFormat("SubType[%u/top]: %s NULL\n", it->first, subTypeToText(it->first).c_str());
	}
	str += "</table>\n";
	std::map<PsyContext, uint64>::iterator i, e;
	uint32 count = 0;
	for (i = activeContexts.begin(), e = activeContexts.end(); i != e; i++) {
		if (((PsyContext)i->first).isValid()) {
			str += utils::StringFormat("Active context: %s (%s)<br>\n",
				((PsyContext)i->first).toString().c_str(),
				PrintTimeString(i->second).c_str());
			count++;
		}
	}
	count = 0;
	for (i = inactiveContexts.begin(), e = inactiveContexts.end(); i != e; i++) {
		if (((PsyContext)i->first).isValid()) {
			str += utils::StringFormat("Inactive context: %s (%s)<br>\n",
				((PsyContext)i->first).toString().c_str(),
				PrintTimeString(i->second).c_str());
			count++;
		}
	}
	return str;
}

std::string SubscriptionEngine::activeContextsXML() {
	if (!activeContexts.size())
		return "<contexts count=\"0\" />\n";
	std::string str;
	std::map<PsyContext, uint64>::iterator it, itEnd;
	uint32 count = 0;
	for (it = activeContexts.begin(), itEnd = activeContexts.end(); it!=itEnd; it++) {
		if (((PsyContext)it->first).isValid()) {
			str += utils::StringFormat("<active context=\"%s\" time=\"%llu\" />\n", ((PsyContext)it->first).toString().c_str(), it->second);
			count++;
		}
	}
	for (it = inactiveContexts.begin(), itEnd = inactiveContexts.end(); it!=itEnd; it++) {
		if (((PsyContext)it->first).isValid()) {
			str += utils::StringFormat("<inactive context=\"%s\" time=\"%llu\" />\n", ((PsyContext)it->first).toString().c_str(), it->second);
			count++;
		}
	}
	return utils::StringFormat("<contexts count=\"%u\">\n%s</contexts>\n", count, str.c_str());
}

std::string SubscriptionEngine::toXML() {
	if (!subTriggers.size())
		return "<subscriptions count=\"0\" />\n";
	std::string str;
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	uint32 count = 0;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++) {
		if (it->second) {
			str += it->second->toXML();
			count++;
		}
	}
	return utils::StringFormat("<subscriptions count=\"%u\">\n%s</subscriptions>\n", count, str.c_str());
}

TriggerSpec* SubscriptionEngine::findTrigger(uint32 compID, const char* name) {
	if (!subTriggers.size())
		return NULL;
	TriggerSpec* triggerSpec;
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++) {
		if ((it->second) && (triggerSpec = it->second->findTrigger(compID, name)))
			return triggerSpec;
	}
	return NULL;
}

std::list<PostSpec*>* SubscriptionEngine::findPosts(uint32 compID, const char* name) {
	if (!subTriggers.size())
		return NULL;
	std::list<PostSpec*>* postSpecs;
	std::map<uint16, SubTriggers*>::iterator it, itEnd;
	for (it = subTriggers.begin(), itEnd = subTriggers.end(); it!=itEnd; it++) {
		if ((it->second) && (postSpecs = it->second->findPosts(compID, name)))
			return postSpecs;
	}
	return NULL;
}



bool SubscriptionEngine::setContextActive(const PsyContext& activeContext, const PsyContext& inactiveContext) {
	uint64 now = GetTimeNow();
	activeContexts[activeContext] = now;
	inactiveContexts.erase(activeContext);
	if (inactiveContext != NOCONTEXT) {
		activeContexts.erase(inactiveContext);
		inactiveContexts[inactiveContext] = now;
	}
	return true;
}

bool SubscriptionEngine::setContextInactive(const PsyContext& inactiveContext) {
	if (inactiveContext != NOCONTEXT) {
		uint64 now = GetTimeNow();
		activeContexts.erase(inactiveContext);
		inactiveContexts[inactiveContext] = now;
	}
	return true;
}

bool SubscriptionEngine::processContext(const PsyContext& context) {
	PsyContext inactive;
	std::map<PsyContext, uint64>::iterator it = activeContexts.begin(), itEnd = activeContexts.end();
	while (it != itEnd) {
		switch (context.compare(it->first)) {
			case CONTEXT_OTHER_ROOT:
				// ignore and add later if no other matches
				break;
			case CONTEXT_EQUAL:
			case CONTEXT_INCLUDED:
				// we have it in the list already, no change
				return true;
			case CONTEXT_SUPERSEDES:
				// old context going out, new one coming in
				inactive = it->first;
				setContextActive(context, inactive);
				node->registerNewActiveContext(context, inactive);
				return true;
			default:
				// keep searching...
				break;
		}
		it++;
	}
	// This is a new root context, add and register
	activeContexts[context] = GetTimeNow();
	node->registerNewActiveContext(context);
	return true;
}

bool SubscriptionEngine::isContextActive(const PsyContext& context) {
	std::map<PsyContext, uint64>::iterator it = activeContexts.begin(), itEnd = activeContexts.end();
	while (it != itEnd) {
		if (((PsyContext)it->first).matches(context))
			return true;
		it++;
	}
	return false;
}


// ******************** Get Triggers ***********************

std::string SubscriptionEngine::printFriendlyTriggersForMsg(DataMessage* msg) {

	std::string str = utils::StringFormat("Checking msg type %s (%s) against root type %u\n",
		typeToText(msg->getType()).c_str(), msg->getType().toString().c_str(), msg->getType()[0]);

	SubTriggers* sub;
	if (sub = subTriggers[msg->getType()[0]]) {
		str += utils::StringFormat("Found matching sub for root type %u\n", msg->getType()[0]);
		str += sub->printFriendlyTriggersForMsg(msg);
	}
	else
		str += utils::StringFormat("Didn't find matching sub for root type %u\n", msg->getType()[0]);

	if (sub = subTriggers[0xFFFF]) {
		str += utils::StringFormat("Found matching sub for wildcard root type\n");
		str += sub->printFriendlyTriggersForMsg(msg);
	}
		
	return str;
}

std::list<TriggerSpec*>* SubscriptionEngine::getTriggersForMsg(DataMessage* msg) {
	// Compatibility wrapper that returns a heap-allocated list. The message
	// routing hot path uses the fill-in-place overload below instead, which
	// avoids a per-message heap allocation (and a latent leak that occurred
	// when a message matched no triggers).
	std::list<TriggerSpec*>* triggers = new std::list<TriggerSpec*>;
	getTriggersForMsg(msg, *triggers);
	return triggers;
}

void SubscriptionEngine::getTriggersForMsg(DataMessage* msg, std::list<TriggerSpec*>& triggersRef) {
	std::list<TriggerSpec*>* triggers = &triggersRef;
	triggers->clear();
	if (!msg) {
		LogPrint(0,0,0,"No msg in getTriggersForMsg...");
		return;
	}

	// wait for any writes to finish;
	writeMutex.enter();
	entranceMutex.enter();
	currentReaders++;
	entranceMutex.leave();
	writeMutex.leave();

	SubTriggers* sub;
	std::map<uint16, SubTriggers*>::iterator si;
	// find() rather than operator[]: the latter inserts a NULL entry on a miss,
	// mutating the map under the shared read lock (racing other routing threads).
	if ((si = subTriggers.find(msg->getType()[0])) != subTriggers.end() && (sub = si->second))
		sub->getTriggersForMsg(msg, triggers);
	if ((si = subTriggers.find(0xFFFF)) != subTriggers.end() && (sub = si->second))
		sub->getTriggersForMsg(msg, triggers);

	// release read access
	entranceMutex.enter();
	currentReaders--;
	entranceMutex.leave();

	if (!triggers->size()) {
		//LogPrint(0,0,0,"No triggers in getTriggersForMsg...");
		return;
	}

	uint32 filtertest = 0;

	bool shouldErase;
	TriggerSpec* trigger;
	FilterSpec* filter;
	PsyContext msgContext;
	std::list<TriggerSpec*>::iterator it = triggers->begin(), itEnd = triggers->end();
	while (it != itEnd) {
		trigger = (*it);
		// Filter triggers by active contexts
		if (!isContextActive(trigger->context)) {
			filtertest = 1;
			triggers->erase(it++);
			continue;
		}
		// Filter triggers by other fields
		if ((trigger->from) && (trigger->from != msg->getFrom())) {
			filtertest = 2;
			triggers->erase(it++);
			continue;
		}
		if ((trigger->to) && (trigger->to != msg->getTo())) {
			filtertest = 3;
			triggers->erase(it++);
			continue;
		}
		if ((trigger->tag) && (trigger->tag != msg->getTag())) {
			filtertest = 4;
			triggers->erase(it++);
			continue;
		}
		if (trigger->triggerContext.isValid()) {
			// msgContext = (PsyContext*)msg->getData("Context", size);
			msgContext = msg->getContextChange();
			if (!msgContext.isValid() || (!trigger->triggerContext.matches(msgContext))) {
				filtertest = 5;
				triggers->erase(it++);
				continue;
			}
		}

		float64 fvalue;
		std::string svalue;
		shouldErase = false;
		for (uint32 n=0; n<trigger->filterCount; n++) {
			if (filter = trigger->getFilterSpec(n)) {
				switch (filter->filter) {
					case TRIGGERFILTER_HASKEY:
						// Any entry type counts (hasKey); getData() would only see
						// raw binary entries and miss string/number entries.
						if (!msg->hasKey(filter->key)) {
							filtertest = 6;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_EQUALS:
						svalue = msg->getAsString(filter->key);
						if (strcmp(svalue.c_str(), filter->value) != 0) {
							filtertest = 7;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_NOTEQUALS:
						svalue = msg->getAsString(filter->key);
						if (strcmp(svalue.c_str(), filter->value) == 0) {
							filtertest = 8;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_EQUALSNUMERIC:
						fvalue = msg->getAsFloat(filter->key);
						if (utils::CompareFloats(fvalue, utils::Ascii2Float64(filter->value)) != 0) {
							filtertest = 9;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_NOTEQUALSNUMERIC:
						fvalue = msg->getAsFloat(filter->key);
						if (utils::CompareFloats(fvalue, utils::Ascii2Float64(filter->value)) == 0) {
							filtertest = 10;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_GREATERTHAN:
						fvalue = msg->getAsFloat(filter->key);
						if (utils::CompareFloats(fvalue, utils::Ascii2Float64(filter->value)) < 0) {
							filtertest = 11;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_LESSTHAN:
						fvalue = msg->getAsFloat(filter->key);
						if (utils::CompareFloats(fvalue, utils::Ascii2Float64(filter->value)) > 0) {
							filtertest = 12;
							shouldErase = true;
						}
						break;
					case TRIGGERFILTER_MAXAGE:
						if (filter->time && (GetTimeAgeMS(msg->getCreatedTime()) > filter->time)) {
							filtertest = 13;
							shouldErase = true;
						}
						break;
					default:
						break;
				}
			}
		}
		if (shouldErase) {
			triggers->erase(it++);
			continue;
		}

		it++;
	}

	if (filtertest && !triggers->size()) {
		LogPrint(0,0,0,"Msg filtered in getTriggersForMsg, filter %u...", filtertest);
	}
}

}
