//------------------------------------------------------------------------------
// <copyright file="LqrSMACQueueWorker.h" company="LiquidRobotics">
//
// Copyright (c) 2014 Liquid Robotics, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
//
// This software is associated with a release.  Liquid Robotics makes no
// representation that the design, interfaces or component distribution will
// remain unchanged in future releases.
//
// </copyright>
//
// <summary>
//   Declaration of LqrSMACQueueWorker, a worker thread that enqueues work items
//   and then fires events or exposes member functions to handle the work items.
// </summary>
//
// <owner>Mike Cookson</owner>
//------------------------------------------------------------------------------

#ifndef __LQRSMACQUEUEWORKER_H__
#define __LQRSMACQUEUEWORKER_H__

#include "LqrSMACDeclareException.h"
#include "LqrSMACLockableConcept.h"
#include "LqrSMACSharedLib.h"
#include "LqrSMACWaitableThreadSafeQueue.h"
#include <boost/signals2.hpp>
#include <boost/thread.hpp>
#include <boost/thread/locks.hpp>
#include <boost/thread/recursive_mutex.hpp>
#include <boost/utility.hpp>

namespace LiquidRobotics {
namespace SMACFramework {

class SHARED_LIB_PUBLIC QueueWorkerBase
: public LockableConceptMutex
, boost::noncopyable
{
public:

	DECLARE_EXCEPTION(AlreadyInitialized, std::runtime_error);
	DECLARE_EXCEPTION(NotInitialized, std::runtime_error);
	DECLARE_EXCEPTION(ShutdownRequested, std::runtime_error);

	typedef boost::signals2::scoped_connection ScopedConnection;
	typedef boost::shared_ptr<ScopedConnection> ScopedConnectionPtr;
	typedef boost::signals2::signal<void (QueueWorkerBase& sender)> ReadTimeoutSignalType;
	typedef boost::signals2::signal<void (QueueWorkerBase& sender)> ShutdownWarningSignalType;
	typedef boost::signals2::signal<void (QueueWorkerBase& sender)> WakeupSignalType;

	QueueWorkerBase();
	virtual ~QueueWorkerBase();

	virtual LockableConceptMutex& GetFiringMutex();
	virtual void Initialize();
	virtual bool IsInitialized();
	virtual bool IsShutdown();
	virtual void Shutdown();
	virtual ScopedConnectionPtr SubscribeReadTimeout(ReadTimeoutSignalType::slot_type slot);
	virtual ScopedConnectionPtr SubscribeShutdownWarning(ShutdownWarningSignalType::slot_type slot);
	virtual ScopedConnectionPtr SubscribeWakeup(WakeupSignalType::slot_type slot);

protected:

	typedef boost::thread Thread;
	typedef boost::shared_ptr<Thread> ThreadPtr;

	enum QueueEntryType
	{
		ET_CLIENT_ITEM,								// item with client object attached was read from queue
		ET_SHUTDOWN,								// item with shutdown request attached was read from queue
		ET_TIMEOUT,									// nothing was read within the specified timeout period
		ET_WAKEUP									// used to generate an event for derived class
	};

	virtual void BackgroundThreadFunc();			// function that receives control from background thread
	virtual bool CheckState(bool isInit=false, bool throwOnError=true);	// checks state flags
	virtual int DetermineReadTimeout();				// number of milliseconds to wait for read ready, <0=wait forever
	virtual void _Shutdown(Lock* pLock = NULL);		// perform shutdown actions

	// reads the next item in the queue (timeout is "readTimeoutMS" in seconds, or wait forever if <0)
	virtual QueueEntryType ReadQueueEntry(int readTimeoutMS) = 0;

	// writes an element in the queue with no client payload
	virtual void WriteQueueEntry(QueueEntryType qeType) = 0;

	virtual void OnAnnounceShutdownWarning();		// announces shutdown
	virtual void OnAnnounceTimeout();				// announces timeout
	virtual void OnAnnounceWakeup();				// announces wakeup event processing

	bool initialized;								// false initially or if shut down, true for valid operation
	bool shutDown;									// false initially, true if shut down
	ThreadPtr thd;									// the thread that was spawned
	ReadTimeoutSignalType readTimeoutSignal;		// signal announces read timeouts
	ShutdownWarningSignalType shutdownWarningSignal; // signal announces that the object is about to shut down
	WakeupSignalType wakeupSignalType;				// signal announces wakeup event
	LockableConceptMutex firingMutex;				// locked when firing events
};

template<typename QELEM>
class QueueWorker : public QueueWorkerBase
{
public:

	typedef boost::signals2::signal<void (QueueWorker<QELEM>& sender,const QELEM& clientItem)> WorkItemProcessedSignalType;
	typedef boost::shared_ptr<QELEM> QELEMPtr;

	QueueWorker(int _queueCapacity = 0)
	: queue(_queueCapacity)
	, initialCapacity(_queueCapacity)
	{
	}

	virtual ~QueueWorker()
	{
	}

	virtual size_t GetCapacity() const
	{
		return queue.GetCapacity();
	}

	virtual size_t CountEnqueued() const
	{
		return queue.Size();
	}

	virtual bool EnqueueWorkItem(const QELEM& qElem, bool throwOnError = true)
	{
		Lock lock(*this);
		if (!CheckState(false, throwOnError)) return false;

		QELEMPtr clientItem;
		clientItem.reset(new QELEM(qElem));

		QQItemPtr newItem;
		newItem.reset(new QQItem(ET_CLIENT_ITEM, clientItem));

		if (initialCapacity)
			queue.Write(newItem);
		else
			queue.WriteExpand(newItem);

		return true;
	}

	virtual bool IsEmpty() const
	{
		return queue.IsEmpty();
	}

	virtual bool IsFull() const
	{
		return queue.IsFull();
	}

	virtual size_t SpaceAvailable() const
	{
		return queue.SpaceAvailable();
	}

	virtual ScopedConnectionPtr SubscribeWorkItemProcessed(typename WorkItemProcessedSignalType::slot_type slot)
	{
		ScopedConnectionPtr result;
		result.reset( new ScopedConnection(workItemProcessedSignal.connect(slot)) );
		return result;
	}

protected:

	template<typename QE>
	struct QueueItem
	{
		typedef boost::shared_ptr<QE> QELEMPtr;

		QueueItem(QueueEntryType _qeType)
		: qeType(_qeType)
		{
		}

		QueueItem(QueueEntryType _qeType, QELEMPtr _clientItem)
		: qeType(_qeType)
		, clientItem(_clientItem)
		{
		}

		const QueueEntryType qeType;
		QELEMPtr clientItem;
	};

	typedef QueueItem<QELEM> QQItem;
	typedef boost::shared_ptr<QQItem> QQItemPtr;
	typedef WaitableThreadSafeQueue<QQItemPtr> QueueType;

	virtual void OnAnnounceWorkItemProcessed(const QELEM& qElem)
	{
		LockableConceptMutex::Lock lock(GetFiringMutex());
		workItemProcessedSignal(*this, qElem);
	}

	virtual QueueEntryType ReadQueueEntry(int readTimeoutMS)
	{
		QQItemPtr readItem;

		if (readTimeoutMS < 0)
			readItem = queue.Read();

		else
		{
			typename QueueType::Lock lock(queue, boost::defer_lock);

			if (queue.AwaitReadable((unsigned) readTimeoutMS, &lock, false))
				readItem = queue.Read();
			else
				readItem.reset(new QQItem(ET_TIMEOUT));
		}

		if (readItem->qeType == ET_CLIENT_ITEM)
			OnAnnounceWorkItemProcessed(*readItem->clientItem);

		return readItem->qeType;
	}

	virtual void WriteQueueEntry(QueueEntryType qeType)
	{
		QQItemPtr newItem;
		newItem.reset(new QQItem(qeType));

		if (initialCapacity)
			queue.Write(newItem);
		else
			queue.WriteExpand(newItem);
	}

private:

	QueueType queue;
	const int initialCapacity;
	WorkItemProcessedSignalType workItemProcessedSignal;
};

}
}

#endif
