#include "WorkerMessageQueueControl.h"
#include <sstream>

#include "Logger.h"
#include "DaemonConfigs.h"
#include "TransferStatDataList.h"
#include "CacheRequestList.h"

#define MESSAGE_QUEUE_KEY 642656000

/// class CMessageQueueJob

CMessageQueueJob::CMessageQueueJob()
{
}

CMessageQueueJob::~CMessageQueueJob()
{
}

bool CMessageQueueJob::ThreadInit(const sig_atomic_t *sighandle, CIPCMsgQueue& msgqueu, CCacheRequestList *pCacheReqList, CHotContentInfoMap  * photContentinfoMap)
{
	m_objHotContentDetectThread.ThreadInit(sighandle, pCacheReqList, photContentinfoMap);
	m_objQueuePopThread.ThreadInit(sighandle, msgqueu, m_objHotContentDetectThread.GetTransferList());
	
	return true;
}

bool CMessageQueueJob::Start()
{
	if ( m_objHotContentDetectThread.Start() && m_objQueuePopThread.Start() )
		return true;

	return false;
}

/// class CWorkerMessageQueueControl

CWorkerMessageQueueControl::CWorkerMessageQueueControl()
{

}

CWorkerMessageQueueControl::~CWorkerMessageQueueControl()
{
	m_vecContentQueue.clear();
	m_vecJobTread.clear();
}

pid_t CWorkerMessageQueueControl::Launcher(const sig_atomic_t *sighandle)
{
	Fork();

	if (m_pid == 0)
	{
	
		while (*sighandle == 0)
		{
			// work
			set_ps_display("Worker [Message Queue Control]", false);
			do_work(sighandle);
			
			// signal 받은 경우 wait 불필요
			if(*sighandle == 0)
			{
				pause();
			}
		}

		// httpd 구동 여부에 따라 Detache or Destroy를 결정함
		if(IsProcessRun("httpd") == false)
		{
			LOG(LDBG, "xxxx IPCS QUEUE remove.");
			for( std::vector<CIPCMsgQueue>::size_type i = 0; i < m_vecContentQueue.size(); ++i )
			{
				if( m_vecContentQueue[i].Remove() == false )
				{
					LOG(LWAR, "CIPCMsgQueue remove failed.");
				}
			}
		}

		m_launched = false;
	}

	return m_pid;
}


void CWorkerMessageQueueControl::do_work(const sig_atomic_t *sighandle)
{
	// conf의 값 참조하여 queue 개수 만큼 생성
	int nQueueCount = CDeamonConfig::GetInstance()->GetIPCQueueCount();
	int nMaxIPCQueueSize = CDeamonConfig::GetInstance()->GetMaxIPCQueueSize();
	uint32_t nQueueSize = sizeof(struct Transfer_stat) * nMaxIPCQueueSize / nQueueCount;

	// 재구동 시 기존 정보 있을 경우 삭제 처리함
	m_vecContentQueue.clear();
	m_vecJobTread.clear();

	for( int i = 0; i < nQueueCount; i++)
	{
		// 설정값에 개수를... 
		LOG(LDBG, "********CMsgQueuePopThread SIZE %d ", nQueueSize);

		// Msg Queue 객체 생성 및 vector에 등록
		m_vecContentQueue.push_back(CIPCMsgQueue());
		m_vecJobTread.push_back(CMessageQueueJob());
	}

	// 실제 Queue 를 생성 처리 한다.
	for( std::queue<CIPCMsgQueue>::size_type i = 0; i < m_vecContentQueue.size(); ++i )
	{
		m_vecContentQueue[i].Create(MESSAGE_QUEUE_KEY + i, IPC_CREAT | 0644, "nobody", nQueueSize);		
	}

		
	SetThreadSignal(SIGINT);
	SetThreadSignal(SIGTERM);

	m_objCacheGeneratorThread.ThreadInit(sighandle);

	if (m_objCacheGeneratorThread.Start() == false)
	{
		UnSetThreadSignal(SIGINT);
		UnSetThreadSignal(SIGTERM);
		LOG(LERR, "CacheGenerator Thread start failed.");
		exit(EXIT_FAILURE);
	}

	for (std::queue<CMessageQueueJob>::size_type i = 0; i < m_vecJobTread.size(); ++i)
	{
		m_vecJobTread[i].ThreadInit(sighandle, m_vecContentQueue[i], m_objCacheGeneratorThread.GetReqList(), &m_hotContentinfoMap);
		if (m_vecJobTread[i].Start() == false)
		{
			UnSetThreadSignal(SIGINT);
			UnSetThreadSignal(SIGTERM);
			LOG(LERR, "Pop Threads start failed.");
			exit(EXIT_FAILURE);
		}
	}

	///////////////
	UnSetThreadSignal(SIGINT);
	UnSetThreadSignal(SIGTERM);
}
