//-------------------------------------------------------------------------------------------------
// <copyright file="EventManagerImpl.cpp" company="LiquidRobotics">
//  Copyright (c) Liquid Robotics Corporation, confidential and proprietary.  All rights reserved. 
// </copyright>
//
// <summary>
//  Implementation of the EventManagerImpl class, which creates a publisher-
//  subscriber collection of events compatible with the CORBA-declared IEventManager
//  interface
// </summary>
//
// <owner>Mike Cookson</owner>
//-------------------------------------------------------------------------------------------------

#include "LqrSMACIdl.h"
#include "EventManagerImpl.h"
#include "LqrSMACIObjectBroker-inline.h"
#include "LqrSMACGetErrorDescription.h"
#include <boost/bind.hpp>
#include <boost/foreach.hpp>
#include <ostream>
#include <stdexcept>

using namespace std;
using namespace boost;
using namespace LiquidRobotics::SMAC;
using namespace LiquidRobotics::SMACFramework;

namespace LiquidRobotics {
namespace EventManager {

// Constructor using the ORB only
EventManagerImpl::EventManagerImpl(IObjectBroker& _broker, Logger& _logger, int _maxSubscribers)
: broker(_broker)
, logger(_logger)
, maxSubscribers(_maxSubscribers)
, shutdownQ(*this, _maxSubscribers)
{
	shutdownQ.Initialize();
}

// Destructor
EventManagerImpl::~EventManagerImpl()
{
	Lock lock(mtx);

	while(lookupSize() != 0)
	{
		string ior = lookup.begin()->first;

		lock.unlock();
		DeleteSubscriber(ior.c_str());
		lock.lock();
	}

	lock.unlock();
	shutdownQ.Shutdown();
}

::CORBA::Double EventManagerImpl::Ping()
{
	stringstream msg;
	msg << "EventManager::Ping: ";

	::CORBA::Double result;
	timeval tv;
	int status = gettimeofday(&tv, NULL);
	int errSave = errno;

	if (!status)
	{
		result = tv.tv_usec;
		result /= 1000000.0;
		result += tv.tv_sec;
	}
	else
	{
		msg << "(gettimeofday returned error #" << errSave << " '" << GetErrorDescription(errSave) << "') ";
		result = 0;
	}

	Logger::Locked(logger)->Info << msg.str() << result << endl;
	return result;
}

void EventManagerImpl::AddSubscriber(const char* ior)
{
	AddSubscriber2(ior, "(nil)");
}

void EventManagerImpl::AddSubscriber2(const char* ior, const char* nickName)
{
	const string subrout = "EventManager::AddSubscriber2";

	if (!ior || !*ior)
	{
		Logger::Locked(logger)->Error << subrout << ": invalid 'ior' parameter" << endl;
		ThrowCORBAException2<BadParameterException>();
	}

	if (!nickName)
	{
		Logger::Locked(logger)->Error << subrout << ": invalid 'nickName' parameter" << endl;
		ThrowCORBAException2<BadParameterException>();
	}

	Lock lock(mtx);
	LookupType::iterator foundItem = lookup.find(ior);

	if (foundItem == lookup.end())
	{
		if (lookupSize() >= maxSubscribers)
		{
			Logger::Locked(logger)->Error << subrout << ": subscription count of " << maxSubscribers << " is already filled" << endl;
			ThrowCORBAException<TooManySubscribersException>();
		}

		bool initOK = false;

		try
		{
			SubscriberPtr newSubscriber;
			newSubscriber.reset(new Subscriber(ior, nickName, 200, broker, logger));
			ConnectionTypePtr connectionPtr;
			connectionPtr.reset(new ConnectionType(newSubscriber->ErrorSignal.connect(boost::bind(&EventManagerImpl::ErrorSignalHandler, this, _1))));
			newSubscriber->Initialize();
			initOK = true;
			connections[ior] = connectionPtr;
			lookup[ior] = newSubscriber;
			Logger::Locked(logger)->Info << subrout << "(" << nickName << " = " << ior << ") -> " << connections.size() << " connections" << endl;
		}
		catch(...)
		{
			if (initOK)
			{
				LookupConnectionType::iterator itConn = connections.find(ior);

				if (itConn != connections.end())
					connections.erase(itConn);

				LookupType::iterator it = lookup.find(ior);

				if (it != lookup.end())
				{
					it->second->Shutdown();
					lookup.erase(it);
				}
			}

			Logger::Locked(logger)->Error << subrout << " caught exception" << endl;
			ThrowCORBAException<TooManySubscribersException>();
		}
	}
	else
	{
		Logger::Locked(logger)->Error << subrout << ": ignoring duplicate subscription of IOR " << ior << endl;
	}
}

/*::CORBA::Long EventManagerImpl::AddFilter(const char* ior, const ::CosNotification::FixedEventHeader& header, const ::CosNotification::FilterableEventBody& filter)
{
	return 0;
}*/

void EventManagerImpl::DeleteSubscriber(const char* ior)
{
	if (!ior || !*ior)
	{
		Logger::Locked(logger)->Error << "EventManagerImpl::DeleteSubscriber: invalid 'ior' parameter" << endl;
		ThrowCORBAException2<BadParameterException>();
	}

	Lock mutexLock(mtx);
	LookupConnectionType::iterator foundConnection = connections.find(ior);

	if (foundConnection != connections.end())
		connections.erase(foundConnection);

	LookupType::iterator foundItem = lookup.find(ior);
	string nick;

	if (foundItem != lookup.end())
	{
		nick = foundItem->second->Nick();
		foundItem->second->Shutdown();
		lookup.erase(foundItem);
	}
	else
		nick = ior;

	Logger::Locked(logger)->Info << "EventManagerImpl::DeleteSubscriber(" << nick << ") -> " << connections.size() << " connections" << endl;
}

/*void EventManagerImpl::RemoveFilter(const char* ior, ::CORBA::Long filterKey)
{
}*/

void EventManagerImpl::SetCustomFilter(const char* ior, const ::LiquidRobotics::SMAC::EventManagerCustomFilter& filter)
{
	if (!ior || !*ior)
	{
		Logger::Locked(logger)->Error << "EventManagerImpl::SetCustomFilter: invalid 'ior' parameter" << endl;
		ThrowCORBAException2<BadParameterException>();
	}

	Lock mutexLock(mtx);
	LookupType::iterator foundItem = lookup.find(ior);
	string nick = ior;

	if (foundItem == lookup.end())
	{
		Logger::Locked(logger)->Error << "EventManagerImpl::SetCustomFilter: no match for ior " << ior << endl;
		ThrowCORBAException2<BadParameterException>();
	}
	else
		nick = foundItem->second->Nick();

	stringstream ss;

	ss << "EventManagerImpl::SetCustomFilter for " << nick
	   << " filter { enabled=" << filter.enabled;

	const _CORBA_ULong ateLength = filter.aliasToEventName.length();

	ss << " aliasToEventName={";

	for (_CORBA_ULong idx = 0; (idx+1) < ateLength; idx += 2)
		ss << " \"" << filter.aliasToEventName[idx] << "\"<-\"" << filter.aliasToEventName[idx+1] << "\"";

	ss << " } nameList={";

	const _CORBA_ULong nlLength = filter.nameList.length();

	for (_CORBA_ULong idx = 0; idx < nlLength; ++idx)
		ss << " \"" << filter.nameList[idx] << "\"";

	ss << " } typeAndDomainList={";

	const _CORBA_ULong tadLength = filter.typeAndDomainList.length();

	for (_CORBA_ULong idx = 0; (idx+1) < tadLength; idx += 2)
		ss << " \"" << filter.typeAndDomainList[idx] << "\"/\"" << filter.typeAndDomainList[idx+1] << "\"";

	ss << "} }";

	Logger::Locked(logger)->Info << ss.str() << endl;

	SubscriberPtr sub = foundItem->second;

	sub->SetCustomFilter(filter);
}

void EventManagerImpl::offer_change(const ::CosNotification::EventTypeSeq& added, const ::CosNotification::EventTypeSeq& removed)
{
	// no handling for this
}

inline std::ostream& operator<<(std::ostream& stm, const ::CosNotification::StructuredEvent& notification)
{
	stm << "event " << notification.header.fixed_header.event_name
	    << " type " << notification.header.fixed_header.event_type.type_name
	    << "/" << notification.header.fixed_header.event_type.domain_name;

	return stm;
}

void EventManagerImpl::push_structured_event(const ::CosNotification::StructuredEvent& notification)
{
	Lock lock(mtx);
	const int luSize = lookupSize();

	stringstream errors;
	stringstream publishedTo;
	unsigned numSubscribers = 0;

	// if there are any subscribers ...
	if (luSize != 0)
	{
		// makes a new shared pointer to avoid duplicating the structured event so much
		QueueItem newItem = QueueItem(new StructuredEvent(notification));

		// iterate through the map and queue a copy of the shared pointer on each subscriber,
		// let them contact subscribers asynchronously
		BOOST_FOREACH(LookupType::value_type val, lookup)
		{
			const string& nick = val.second->Nick();

			try
			{
				if (val.second->Queue(newItem))
				{
					if (!publishedTo.tellp())
						publishedTo << ":";

					publishedTo << " " << nick;
					++numSubscribers;
				}
			}
			catch(Subscriber::QueueTooLong& qtl)		// automatically disconnect if too many events are queued up
			{
				if (!errors.tellp())
					errors << notification << ": event queue too long for the following (shutting down):";

				errors << " " << nick;
				shutdownQ.QueueShutdown(val.first.c_str());
			}
			catch(...)					// not allowed to throw exceptions to subscriber, so catch all
			{
			}
		}
	}

	Logger::Locked(logger)->Info
		<< "Received " << notification << " and published to " << numSubscribers << " subscribers"
		<< publishedTo.str() << endl;

	if (errors.tellp())
		Logger::Locked(logger)->Error << errors.str() << endl;
}

void EventManagerImpl::disconnect_structured_push_consumer()
{
	// no handling for this
}

void EventManagerImpl::ErrorSignalHandler(const char* ior)
{
	string nick = ior ? ior : "";
	LookupType::iterator foundItem = lookup.find(nick);

	if (foundItem != lookup.end())
		nick = foundItem->second->Nick();

	Logger::Locked(logger)->Error << "Caught error signal for " << nick << ": the queue will be shut down" << endl;
	shutdownQ.QueueShutdown(ior);
}

EventManagerImpl::Subscriber::Subscriber(const char* iorP, const char* nickP, int maxQueueSize, IObjectBroker& _broker, Logger& _logger)
: base(maxQueueSize)
, broker(_broker)
, logger(_logger)
, ior(iorP)
, nick(nickP)
{
	ets.ClearEventFilter();
	ets.SetEventFilterEnabled(false);
	ets.RemoveAllAliasesForEvents();
}

EventManagerImpl::Subscriber::~Subscriber()
{
}

/*long EventManagerImpl::Subscriber::AddFilter(const ::CosNotification::FixedEventHeader& header, const ::CosNotification::FilterableEventBody& filter)
{
	return 0;
}*/

const char* EventManagerImpl::Subscriber::IOR() const
{
	return ior.c_str();
}

const EventManagerImpl::Subscriber::String& EventManagerImpl::Subscriber::Nick() const
{
	return nick;
}

/*void EventManagerImpl::Subscriber::RemoveFilter(long filterKey)
{
}*/

void EventManagerImpl::Subscriber::SetCustomFilter(const ::LiquidRobotics::SMAC::EventManagerCustomFilter& filter)
{
	ets.SetCustomFilter(filter);
}

bool EventManagerImpl::Subscriber::Queue(QueueItem& eventToFire)
{
	typedef LiquidRobotics::EventLib::EventType EventType;
	bool filtered = false;

	if (ets.EventFilterEnabled())
	{
		string eventName = eventToFire->header.fixed_header.event_name.in();

		EventType eventType(eventToFire->header.fixed_header.event_type.type_name.in(),
							eventToFire->header.fixed_header.event_type.domain_name.in());

		filtered = !ets.EventNameEnabled(eventName) && !ets.EventTypeEnabled(eventType);
	}

	const bool result = !filtered;

	if (result)
		DoEnqueue(eventToFire);

	return result;
}

void EventManagerImpl::Subscriber::BackgroundTask()
{
	StructuredPushConsumer_var push;
	bool error = false;

	while(1)
	{
		bool setError = false;

		// check before waiting if there is a shutdown request
		if (shutdownRequest)
			break;

		// await queue readable and return with qLock locked when it is
		ItemQueue::Lock qLock(queue, boost::defer_lock);
		queue.AwaitReadable(&qLock);

		// terminate the loop if there was a shutdown while waiting
		if (shutdownRequest)
			break;

		// read the item from the queue and unlock the lock
		boost::shared_ptr<CosNotification::StructuredEvent> item = queue.Read();
		qLock.unlock();

		string iorCopy = ior;

		// if a null item was pushed, that counts as a shutdown request
		if (!item)
			shutdownRequest = true;

		// if no error so far ..
		if (!setError && !error && !shutdownRequest)
		{
			try
			{
				// get a new copy of the consumer if the pointer is nil
				if (push.in() == StructuredPushConsumer::_nil())
					push = broker.GetObjectFromStringizedName<StructuredPushConsumer_var,StructuredPushConsumer>(ior.c_str());

				// push the event to the consumer
				push->push_structured_event(*item);
			}
			catch(std::exception& ex)
			{
				Logger::Locked(logger)->Error << "Caught " << typeid(ex).name() << " (" << ex.what() << ") when forwarding event to " << ior << endl;
				setError = true;
			}
			catch(...)
			{
				Logger::Locked(logger)->Error << "Caught exception when forwarding event to " << ior << endl;
				setError = true;
			}
		}
		else if (!shutdownRequest)
			Logger::Locked(logger)->Info << "Ignoring message for " << ior << " since errors were identified" << endl;

		// signal an error if an exception was thrown
		if (setError && !error)
		{
			error = true;
			ErrorSignal(iorCopy.c_str());
		}
	}
}

void EventManagerImpl::ShutdownQueue::WriteShutdownItem()
{
	queue.Write("");
}

void EventManagerImpl::Subscriber::WriteShutdownItem()
{
	queue.Write(QueueItem());
}


EventManagerImpl::ShutdownQueue::ShutdownQueue(EventManagerImpl& eventMgrP, int maxQueueSizeP)
: base(maxQueueSizeP)
, eventMgr(eventMgrP)
{
}

EventManagerImpl::ShutdownQueue::~ShutdownQueue()
{
}

void EventManagerImpl::ShutdownQueue::QueueShutdown(const char* ior)
{
	queue.Write(ior);
}

void EventManagerImpl::ShutdownQueue::BackgroundTask()
{
	while(1)
	{
		Lock lock(mtx);

		if (shutdownRequest)
			break;

		// await queue readable and return with qLock locked when it is
		ItemQueue::Lock qLock(queue, boost::defer_lock);
		queue.AwaitReadable(&qLock);

		QueueItem item = queue.Read();
		qLock.unlock();

		if (!item.length())
			shutdownRequest = true;

		if (shutdownRequest)
			break;

		try
		{
			eventMgr.DeleteSubscriber(item.c_str());
		}
		catch(...)
		{
		}
	}
}

}
}
