//-------------------------------------------------------------------------------
// <copyright file="LqrSMACWaitableThreadSafeQueue.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>
//   Implementation of the WaitableThreadSafeQueue<T> class, a generic FIFO queue
//   that is thread safe and which implements methods of pending synchronously on
//   read or write.
// </summary>
//
// <owner>Mike Cookson</owner>
//-------------------------------------------------------------------------------

#ifndef __LqrSMACWaitableThreadSafeQueue_H__
#define __LqrSMACWaitableThreadSafeQueue_H__

#include "LqrSMACSemaphore.h"
#include "LqrSMACThreadSafeQueue.h"
#include "LqrSMACUptime.h"

namespace LiquidRobotics {
namespace SMACFramework {

namespace WaitableThreadSafeQueueTools
{
	SHARED_LIB_PUBLIC bool AwaitSem(float startTime, unsigned millisecTimeout, LiquidRobotics::SMACFramework::Semaphore& sem);
}

template<typename T>
class SHARED_LIB_PUBLIC WaitableThreadSafeQueue : public ThreadSafeQueue<T>
{
public:

	typedef ThreadSafeQueue<T> Base;
	typedef typename Base::Lock Lock;
	typedef boost::shared_ptr<Lock> LockPtr;
	typedef typename Base::size_type size_type;
	typedef boost::shared_ptr<WaitableThreadSafeQueue> Ptr;

	WaitableThreadSafeQueue(size_type initialCapacity = 0)
	: Base(initialCapacity)
	{
		UpdateWriteStatus();
	}

	virtual void AwaitReadable(Lock* callersLock = NULL) const
	{
		LockPtr lockPtr;

		if (!callersLock)
		{
			lockPtr.reset(new Lock(const_cast< WaitableThreadSafeQueue<T>& >(*this)));
			lockPtr->unlock();
			callersLock = lockPtr.get();
		}

		bool readable = false;

		while(!readable)
		{
			callersLock->lock();

			if (!Base::IsEmpty())
				readable = true;
			else
			{
				UpdateReadStatus();
				callersLock->unlock();
				const_cast<Semaphore&>(readSem).Wait();
			}
		}

		UpdateReadStatus();
	}

	virtual bool AwaitReadable(unsigned milliseconds, Lock* callersLock = NULL, bool throwOnError = true) const
	{
		LockPtr lockPtr;

		if (!callersLock)
		{
			lockPtr.reset(new Lock(const_cast< WaitableThreadSafeQueue<T>& >(*this)));
			lockPtr->unlock();
			callersLock = lockPtr.get();
		}

		const float startTime = GetSystemUpTimeSeconds();
		bool readable = false;
		bool timeout = false;

		do
		{
			callersLock->lock();

			if (!IsEmpty())
			{
				readable = true;
				UpdateReadStatus();
			}

			else
			{
				UpdateReadStatus();
				callersLock->unlock();
				timeout = !WaitableThreadSafeQueueTools::AwaitSem(startTime, milliseconds, const_cast<Semaphore&>(readSem));
			}

		} while(!readable && !timeout);

		if (timeout && throwOnError)
			throw QueueBase::ReadTimeout(QueueBase::timeoutOnAwaitReadableText);

		return readable;
	}

	virtual void AwaitWriteable(Lock* callersLock = NULL) const
	{
		LockPtr lockPtr;

		if (!callersLock)
		{
			lockPtr.reset(new Lock(const_cast< WaitableThreadSafeQueue<T>& >(*this)));
			lockPtr->unlock();
			callersLock = lockPtr.get();
		}

		bool writeable = false;

		while(!writeable)
		{
			callersLock->lock();

			if (!IsFull())
				writeable = true;
			else
			{
				UpdateWriteStatus();
				callersLock->unlock();
				const_cast<Semaphore&>(writeSem).Wait();
			}
		}

		UpdateWriteStatus();
	}

	virtual bool AwaitWriteable(unsigned milliseconds, Lock* callersLock = NULL, bool throwOnError = true) const
	{
		LockPtr lockPtr;

		if (!callersLock)
		{
			lockPtr.reset(new Lock(const_cast< WaitableThreadSafeQueue<T>& >(*this)));
			lockPtr->unlock();
			callersLock = lockPtr.get();
		}

		const float startTime = GetSystemUpTimeSeconds();
		bool writeable = false;
		bool timeout = false;

		do
		{
			callersLock->lock();
			writeable = !IsFull();

			if (writeable)
				UpdateWriteStatus();

			else
			{
				UpdateWriteStatus();
				callersLock->unlock();
				timeout = !WaitableThreadSafeQueueTools::AwaitSem(startTime, milliseconds, const_cast<Semaphore&>(writeSem));
			}

		} while(!writeable && !timeout);

		if (timeout && throwOnError)
			throw QueueBase::WriteTimeout(QueueBase::timeoutOnAwaitWriteableText);

		return writeable;
	}

	virtual bool IsEmpty() const
	{
		return Base::IsEmpty();
	}

	virtual bool IsFull() const
	{
		return Base::IsFull();
	}

	virtual void SetCapacity(size_type newCapacity)
	{
		Lock lock(*this);
		Base::SetCapacity(newCapacity);
		UpdateWriteStatus();
	}

	virtual void Clear()
	{
		Lock lock(*this);
		Base::Clear();
		UpdateReadWriteStatus();
	}

	virtual T Read()
	{
		Lock lock(*this, boost::defer_lock);

		AwaitReadable(&lock);
		T result = Base::Read();

		UpdateReadWriteStatus();
		return result;
	}

	virtual T Read(unsigned milliseconds)
	{
		Lock lock(*this, boost::defer_lock);

		AwaitReadable(milliseconds, &lock);
		T result = Base::Read();

		UpdateReadWriteStatus();
		return result;
	}

	virtual bool ReadTest(T& result)
	{
		Lock lock(*this);
		bool retval = Base::ReadTest(result);

		UpdateReadWriteStatus();
		return retval;
	}

	virtual bool ReadTest(T& result, unsigned milliseconds)
	{
		Lock lock(*this, boost::defer_lock);
		bool success = AwaitReadable(milliseconds, &lock, false);

		if (success)
		{
			result = Base::Read();
			UpdateReadWriteStatus();
		}

		return success;
	}

	virtual bool Write(const T& newItem, bool throwOnError = true)
	{
		Lock lock(*this, boost::defer_lock);

		AwaitWriteable(&lock);
		Base::Write(newItem, throwOnError);
		UpdateReadWriteStatus();
		return true;
	}

	virtual bool Write(const T& newItem, unsigned milliseconds, bool throwOnError = true)
	{
		Lock lock(*this, boost::defer_lock);
		bool success = AwaitWriteable(milliseconds, &lock, throwOnError);

		if (success)
		{
			Base::Write(newItem);
			UpdateReadWriteStatus();
		}

		return success;
	}

	virtual void WriteExpand(const T& newItem)
	{
		Lock lock(*this);

		Base::WriteExpand(newItem);
		UpdateReadWriteStatus();
	}

protected:

	void UpdateReadWriteStatus() const
	{
		UpdateReadStatus();
		UpdateWriteStatus();
	}

	void UpdateReadStatus() const		// called when the mutex is already locked
	{
		const_cast<Semaphore&>(readSem).SetCount(IsEmpty() ? 0 : 1);
	}

	void UpdateWriteStatus() const		// called when the mutex is already locked
	{
		const_cast<Semaphore&>(writeSem).SetCount(Base::SpaceAvailable() ? 1 : 0);
	}

	static bool AwaitSem(const float startTime, float& timeRemaining, Semaphore& sem);

	Semaphore readSem;
	Semaphore writeSem;
};

}
}


#endif /* LQRSMACWAITABLETHREADSAFEQUEUE_H_ */
