//---------------------------------------------------------------------------------------------------------------------
// pcqueue.hpp
//
// Asker's product consumer queue
//---------------------------------------------------------------------------------------------------------------------
#pragma once

#include <iostream>
#include <vector>
#include <pthread.h>

#include "errno.h"

#include "common.hpp"

using namespace std;
//---------------------------------------------------------------------------------------------------------------------
//---------------------------------------------------------------------------------------------------------------------
//---------------------------------------------------------------------------------------------------------------------
template <class T>
class CPCQueue
{
private:
	pthread_mutex_t ProducerLock;
	pthread_mutex_t ConsumerLock;

	vector<T*> ProducerQueue;
	vector<T*> ConsumerQueue;

private:
	bool LockProducer()
	{
		int _nError = pthread_mutex_lock( &ProducerLock );
		if ( _nError )
		{
			cout << "error in lock producer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in lock producer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}
	
		return true;
	}

	bool UnlockProducer()
	{
		int _nError = pthread_mutex_unlock( &ProducerLock );
		if ( _nError )
		{
			cout << "error in unlock producer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in unlock producer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}
	
		return true;
	}

	bool LockConsumer()
	{
		int _nError = pthread_mutex_lock( &ConsumerLock );
		if ( _nError )
		{
			cout << "error in lock consumer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in lock consumer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}

		return true;
	}

	bool UnlockConsumer()
	{
		int _nError = pthread_mutex_unlock( &ConsumerLock );
		if ( _nError )
		{
			cout << "error in unlock consumer. return : " << _nError << ", err :" << strerror(errno) << endl;
			formatlog( LOG_CRIT, "error in unlock consumer. return(%d) ErrNo(%d) ErrMsg(%s)", _nError, errno, strerror(errno));
			return false;
		}

		return true;
	}


public:
	CPCQueue(){}
	virtual ~CPCQueue()
	{
		for ( class vector<T*>::iterator _iter = ProducerQueue.begin(); _iter != ProducerQueue.end(); _iter++)
			delete *_iter;

		for ( class vector<T*>::iterator _iter = ConsumerQueue.begin(); _iter != ConsumerQueue.end(); _iter++)
			delete *_iter;

		ProducerQueue.clear();
		ConsumerQueue.clear();
	}

	bool Init()
	{
		int _ErrorNo = pthread_mutex_init( &ProducerLock, NULL);
		if (_ErrorNo)
			return false;
	
		_ErrorNo = pthread_mutex_init( &ConsumerLock, NULL);
		if (_ErrorNo)
			return false;
	
		return true;
	}

	// Product에 하나를 추가한다
	bool AddProduct( T* _Job )
	{
		if ( false == LockProducer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		ProducerQueue.push_back(_Job);
	 
		if ( false == UnlockProducer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		return true;
	}

	// Consumer 쪽에서 하나씩 빼낸다.
	bool PopFromConsumer(T** _Value)
	{
		*_Value = NULL;

		if ( false == LockConsumer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		if (!ConsumerQueue.empty())
		{
			*_Value = ConsumerQueue.front();
			ConsumerQueue.erase(ConsumerQueue.begin());
		}

		if ( false == UnlockConsumer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		return true;
	}

	// 합할 것이 하나도 없는 경우 false를 반환한다
	bool MoveProducerToConsumer(int& _MoveCount)
	{
		if ( false == LockConsumer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		if ( false == LockProducer())
		{
			// lock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Lock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		_MoveCount = ProducerQueue.size();
		if (!ProducerQueue.empty())
		{
			for ( class vector<T*>::const_iterator _iter = ProducerQueue.begin(); _iter != ProducerQueue.end(); _iter++)
				ConsumerQueue.push_back(*_iter);
	
			ProducerQueue.clear();
		}

		if ( false == UnlockProducer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		if ( false == UnlockConsumer())
		{
			// unlock 실패
			formatlog( LOG_CRIT, "CPCQueue::%s(%d) Unlock Fail. fatal issue.", __func__, __LINE__);
			return false;
		}

		return true;
	}

	int GetConsumerSize() const
	{
		return ConsumerQueue.size();
	}

};
//---------------------------------------------------------------------------------------------------------------------

class CDataThread: public CUbiThread
{
private:
	CPCQueue<COnePacket> PCQueue;
	int ThreadIndex;

	CURL* ctx;

public:
	bool Terminated;

public:
	CDataThread();
	virtual ~CDataThread();

	bool init();
	void SetIndex(int _Index);
	bool AddProduct( COnePacket* _OnePacket);

	virtual void Run();

private:
	bool DoPacketProcess(COnePacket* _OnePacket);
	CUbiParser* MakeParser(COnePacket* _OnePacket);
}

//---------------------------------------------------------------------------------------------------------------------
CDataThread::CDataThread()
{
}
//---------------------------------------------------------------------------------------------------------------------
CDataThread::~CDataThread()
{
}
//---------------------------------------------------------------------------------------------------------------------
bool CDataThread::init()
{
	return PCQueue.Init();
}
//---------------------------------------------------------------------------------------------------------------------
void CDataThread::SetIndex(int _Index)
{
	ThreadIndex = _Index;
}
//---------------------------------------------------------------------------------------------------------------------
bool CDataThread::AddProduct( COnePacket* _OnePacket )
{
	return PCQueue.AddProduct(_OnePacket);
}
//---------------------------------------------------------------------------------------------------------------------
void CDataThread::Run()
{
	int _Count = 0;
	int _JobCount = 0;
	int _SleepMicro;
	COnePacket* _OnePacket = NULL;

	Terminated = false;
	while(!Terminated)
	{
		try
		{
			// Customer Queue에 Producer Queue의 내용을 가져오게 한다
			if (!PCQueue.MoveProducerToConsumer(_Count))
				formatlog( LOG_FATAL, "CDataThread::%s(%d) %10d Thread Error in Move PCQueue data", __func__, __LINE__, ThreadIndex); //, typeid(T).name());
			else
				formatlog( LOG_INFO, "CDataThread::%s(%d) %10d Thread %2d %3d remain.", __func__, __LINE__, ThreadIndex, _Count, PCQueue.GetConsumerSize());
		}
		catch(exception &_innerException)
		{
			formatlog( LOG_CRIT, "Exception on CDataThread::%s(%d). %2d Thread what(%s)", __func__, __LINE__, ThreadIndex, _innerException.what());
		}

		// Customer Queue를 비울때 까지 작업을 처리 한다
		_JobCount = 0;
		while(true)
		{
			_OnePacket = NULL;

			// 처리 간격에 일정 시간을 둔다
			if (!PCQueue.PopFromConsumer(&_OnePacket))
			{
				formatlog( LOG_FATAL, "CDataThread::%s(%d). %2d Thread Error in Get Data From PCQueue", __func__, __LINE__, ThreadIndex);
				break;
			}

			if ( NULL == _OnePacket)
				break;

			//1 실질적인 패킷 처리!!
			DoPacketProcess(_OnePacket);
			_JobCount++;

			delete _OnePacket;
		}

		usleep(500000 );
	}
}