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

#ifndef __EventManagerImpl_h__
#define __EventManagerImpl_h__

#include <deque>
#include <string>
#include <boost/shared_ptr.hpp>
#include <boost/signals2.hpp>
#include <boost/signals2/connection.hpp>
#include <boost/signals2/signal.hpp>
#include <boost/thread/condition.hpp>
#include <boost/thread/locks.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include "LqrSMACException.h"
#include "LqrSMACIObjectBroker.h"
#include "LqrSMACLogger.h"
#include "LqrSMACWaitableThreadSafeQueue.h"
#include "EventTransceiver_Settings.h"

namespace LiquidRobotics {
namespace EventManager {

class EventManagerImpl
{
public:

	typedef LiquidRobotics::SMACFramework::IObjectBroker IObjectBroker;
	typedef LiquidRobotics::SMACFramework::Logger Logger;

	EventManagerImpl(IObjectBroker& _broker, Logger& _logger, int _maxSubscribers);
	virtual ~EventManagerImpl();

	// IPingable operations
    virtual ::CORBA::Double Ping();

	// IEventManager operations
	virtual void AddSubscriber(const char* ior);
	virtual void AddSubscriber2(const char* ior, const char* nickName);
	virtual void DeleteSubscriber(const char* ior);
    void SetCustomFilter(const char* ior, const ::LiquidRobotics::SMAC::EventManagerCustomFilter& filter);

	// StructuredPushConsumer operations
    virtual void offer_change(const ::CosNotification::EventTypeSeq& added, const ::CosNotification::EventTypeSeq& removed);
    virtual void push_structured_event(const ::CosNotification::StructuredEvent& notification);
    virtual void disconnect_structured_push_consumer();

private:

	void ErrorSignalHandler(const char* ior);

	template<typename __QueueItem, typename __QueueItemParameter> class CommonQueue
	{
	public:

		typedef __QueueItem QueueItem;
		typedef LiquidRobotics::SMACFramework::AlreadyInitialized AlreadyInitialized;
		typedef LiquidRobotics::SMACFramework::Exception Exception;
		typedef LiquidRobotics::SMACFramework::NotInitialized NotInitialized;

		CommonQueue(int maxQueueSizeP)
		: initialized(false)
		, shutdownRequest(false)
		, thd(NULL)
		, queue(maxQueueSizeP)
		{
		}

		virtual ~CommonQueue()
		{
			// --- queue should be shut down in advance Shutdown();
		}

		virtual void Initialize()
		{
			Lock lock(mtx);

			if (initialized)
				throw AlreadyInitialized();

			thd = new Thread(boost::bind(&CommonQueue::BackgroundTask, this));
			initialized = true;
		}

		virtual void Shutdown()
		{
			Lock lock(mtx);

			if (initialized && thd && !shutdownRequest)
			{
				shutdownRequest = true;
				WriteShutdownItem();
				lock.unlock();
				thd->join();
			}
		}

		virtual void WriteShutdownItem()
		{
		}

		class QueueTooLong : public Exception
		{
		public:

			QueueTooLong() throw()
				: Exception("The event queue is too long")
			{
			}
		};

	protected:

		typedef LiquidRobotics::SMACFramework::WaitableThreadSafeQueue<QueueItem> ItemQueue;
		typedef boost::mutex Mutex;
		typedef boost::thread Thread;
		typedef boost::unique_lock<Mutex> Lock;

		bool initialized;						// set to true when the object has been initialized
		bool shutdownRequest;					// set to true when there's a request to shut down the object
		Mutex mtx;								// mutex to protect the state
		Thread* thd;							// thread to run BackgroundTask
		ItemQueue queue;						// queue of IOR's to be processed

		virtual void BackgroundTask() = 0;

		void DoEnqueue(__QueueItemParameter newItem)
		{
			Lock lock(mtx);

			if (!initialized)					throw NotInitialized();
			if (!queue.SpaceAvailable())		throw QueueTooLong();
			if (!shutdownRequest)				queue.Write(newItem);
		}

		// returns a properly-signed queue size
		inline int queueSize() const { return (int) queue.Size(); }
	};

	class ShutdownQueue : public CommonQueue<std::string, const char*>
	{
	public:

		typedef CommonQueue<std::string, const char*> base;

		ShutdownQueue(EventManagerImpl& eventMgrP, int maxQueueSize);
		virtual ~ShutdownQueue();

		void QueueShutdown(const char* ior);

	protected:

		virtual void BackgroundTask();
		virtual void WriteShutdownItem();

	private:

		EventManagerImpl& eventMgr;				// references the event manager instance (i.e. to call Shutdown)
	};

	class Subscriber : public CommonQueue<boost::shared_ptr<CosNotification::StructuredEvent>, boost::shared_ptr<CosNotification::StructuredEvent> >
	{
	public:

		typedef CosNotification::StructuredEvent_var StructuredEvent_var;
		typedef boost::signals2::signal1<void,const char*> ErrorSignalType;
		typedef CommonQueue<boost::shared_ptr<CosNotification::StructuredEvent>, boost::shared_ptr<CosNotification::StructuredEvent> > base;
		typedef CORBA::Object_var Object_var;
		typedef LiquidRobotics::SMACFramework::IObjectBroker IObjectBroker;
		typedef std::string String;

		ErrorSignalType ErrorSignal;

		Subscriber(const char* iorP, const char* nickP, int maxQueueSize, IObjectBroker& _broker, Logger& _logger);
		virtual ~Subscriber();

		const char* IOR() const;
		const String& Nick() const;
		bool Queue(QueueItem& eventToFire);
	    void SetCustomFilter(const ::LiquidRobotics::SMAC::EventManagerCustomFilter& filter);

	protected:

		virtual void BackgroundTask();
		virtual void WriteShutdownItem();

	private:

		typedef LiquidRobotics::EventLib::EventTransceiverSettings EventTransceiverSettings;

		IObjectBroker& broker;				// the object broker
		Logger& logger;						// the logger instance
		String ior;							// the IOR of the client
		String nick;						// nickname of the client
		EventTransceiverSettings ets;		// filtering settings
	};

	typedef boost::shared_ptr<Subscriber> SubscriberPtr;

	inline int lookupSize() const { return (int) lookup.size(); }

	typedef std::map<std::string, SubscriberPtr> LookupType;
	typedef boost::signals2::scoped_connection ConnectionType;
	typedef boost::shared_ptr<ConnectionType> ConnectionTypePtr;
	typedef std::map<std::string, ConnectionTypePtr> LookupConnectionType;
	typedef LiquidRobotics::SMAC::BadParameter BadParameterException;
	typedef LiquidRobotics::SMAC::TooManySubscribers TooManySubscribersException;
	typedef boost::mutex Mutex;
	typedef boost::unique_lock<Mutex> Lock;
	typedef CORBA::Object Object;
	typedef CORBA::Object_var Object_var;
	typedef CosNotification::StructuredEvent StructuredEvent;
	typedef boost::shared_ptr<CosNotification::StructuredEvent> QueueItem;
	typedef CosNotifyComm::StructuredPushConsumer StructuredPushConsumer;
	typedef CosNotifyComm::StructuredPushConsumer_var StructuredPushConsumer_var;

	IObjectBroker& broker;					// the object broker
	Logger& logger;							// logging instance
	int maxSubscribers;						// maximum number of subscribers to allow
	LookupType lookup;						// lookup of each Subscriber by IOR
	LookupConnectionType connections;		// callback connections by IOR to connection var
	Mutex mtx;								// mutex to protect the internal state
	ShutdownQueue shutdownQ;				// queue of IOR's to shut down asynchronously
};

}
}

#endif
