#include "WorkerRecvVmasd.h"

#include "Logger.h"
#include "ProcessRename.h"
#include "ShmManager.h"
#include "FixedQueue.h"
#include "ProtocolVmasd.h"
#include "ProcessConfig.h"

#include <stdlib.h>
#include <signal.h>
#include <pthread.h>
#include <errno.h>
#include <string.h>
#include <arpa/inet.h>


// vmasd broadcast message 를 수신하기 위한 buffer size
// - vmasd 는 Header 352 Byte, Body 640 Byte => 총 992 Byte 전달함  (BroadcastMessage.h 참고)
// - 따라서 향후 확장으로 고려하여 기존 rc_vrcd 9216 Byte  -> 1024 Byte 로 변경함.
#define MAX_SOCKET_BUFFER_SIZE		1024	

// vmasd 로 부터 전달받은 Data 를 파싱 thread 에 전달시 사용할 Queue 의 최대 갯수
#define MAX_DATA_QUEUE_COUNT		3000

// 수신 처리를 담당할 Thread 의 수
#define MAX_THREAD_RECEIVER           10


// vmasd 수신 시작 delay time (sec)
#define RECV_START_DELAY_SEC           5



// 공유메모리에 Data Set, Get 을 담당할 객체에 대한 포인터.
// Main.cpp 에 전역변수로 정의 되어 있다.
extern CShmManager * g_pShmManager;

// FHS 의 vmasd 가 Broadcasting 하는 각 FHS 의 상태 정보를 수신받기 위한 UDP Socket....
// Main.cpp 에 전역변수로 정의 되어 있다.
extern int g_sockVmasdRecv;

// FHS csagentd 가 전송하는 시스템 상태 정보 수신을 위한 TCP Socket
// Main.cpp 에 전역변수로 정의 되어 있다.
extern int g_sockSystemRecv;

// FHS csagentd 가 전송하는 서비스 상태 체크 정보 수신을 위한 TCP Socket
// Main.cpp 에 전역변수로 정의 되어 있다.
extern int g_sockCheckRecv;

// fimngd 등과 같은 외부 모듈과 통신을 통해 정보를 전달하기 위한 TCP Socket 생성.
// Main.cpp 에 전역변수로 정의 되어 있다.
extern int g_sockInfoSend;

// Data 수신부와 이를 파싱하여 처리할 Thread 객체간에 Data 전달을 위한 Queue 객체 생성
CFixedQueue * g_pStatDataQueue = NULL;



// 프로세스 종료 전 처리할 종료 관련 각 작업을 일괄로 처리하기 위한 함수.
void ReadyToExitWorkerRecvVmasd()
{
	// open 된 socket 관련 close 처리
	if( g_sockVmasdRecv != -1 )
		close( g_sockVmasdRecv );

	// Queue 객체 해제
	if( g_pStatDataQueue != NULL )
	{
		CFixedQueue * temp = g_pStatDataQueue;
		
		g_pStatDataQueue = NULL;
		delete temp;		
	}
	
	// Logger 객체 종료 처리.
	CLogger::Exit();

	return;
}

// Worker RecvVmasd 프로세스에 대한 대한 종료 처리 수신시.
static void SignalWorkerRecvVmasdTerminate( int nSignalNumber )
{
	// Signal Number 에 따른 로깅처리.
	if( nSignalNumber == SIGTERM )
	{
		_LOG( LINF, "[RecvVmasd:%d] Process exit by SIGTERM signal. Good Bye..", getpid() );
	}
	else
	{
		_LOG( LWAR, "[RecvVmasd:%d] Process exit by abnormal signal[%d], Good Bye..", getpid(), nSignalNumber );
	}

	ReadyToExitWorkerRecvVmasd();

	// 종료전 잠시 대기
	struct timespec sleep;
	sleep.tv_sec  = 0;
	sleep.tv_nsec = 500000000;  // 0.5 sec
	nanosleep( &sleep, NULL );
	
	exit( EXIT_SUCCESS );
}

// Worker::RecvVmasd Process 의 signal 처리기...
void SetWorkerRecvVmasdSignalHandler()
{
	sigset_t set;
	struct sigaction act;

	memset( &act, 0x00, sizeof( act ) );
	sigfillset( &set );
	sigprocmask( SIG_SETMASK, &set, NULL );
	sigfillset( &act.sa_mask );

	// 무시 처리 signal
	act.sa_handler = SIG_IGN;
	sigaction( SIGPIPE, &act, NULL );	/* desciptor 오류 발생시 Process가 죽는 것은 방지하기 위하여 설정 */
	sigaction( SIGHUP, &act, NULL );	/* process를 기동시킨 관리자의 로그아웃시, 또는 config reload 등의 처리 signal*/
	sigaction( SIGINT, &act, NULL );	/* ^C 키를 누른 경우 받는 신호 => demon 으로 기동되기 땜시 이 신호 못받음 */
	sigaction( SIGQUIT, &act, NULL );	/* 키보드에 의한  Abort 신호 처리 => ? */

	// Worker 프로세스는 자식 프로세스가 존재하지 않으므로 그냥 무시처리함.
	act.sa_handler = SIG_IGN;
	sigaction( SIGCHLD, &act, NULL );

	/* 각종 에러나 사용자의 종료 신호 처리 */
	act.sa_handler = SignalWorkerRecvVmasdTerminate;
	sigaction( SIGTERM, &act, NULL );	/* kill -TERM 에 의한 프로세스 종료시 */

	// 나머지 신호는 default 처리.

	sigemptyset( &set );                  /* 신호 처리기 처리 설정 위한 블록 해제 */
	sigprocmask( SIG_SETMASK, &set, NULL );
}

// 전역 Queue 에 저장된 vmasd 수신 data 를 파싱하여 공유 메모리에 저장 처리를 담당하는 Thread
void * WorkerRecvVmasdSaveThread( void * arg )
{
	// 메인 함수에서 이 스레드가 종료되기를 기다리지 않게 분리
	pthread_detach( pthread_self() );


	_LOG( LINF, "[RecvVmasd:%d] SaveThread: start...", getpid() );


	// Queue pop data 저장 관련 변수 초기화.
	int pop_result;
	char szTempBuffer[MAX_SOCKET_BUFFER_SIZE];

	//struct vmasd_broadcast_message_header * pHeader = (struct vmasd_broadcast_message_header *)szTempBuffer;
	struct vmasd_broadcast_message_body   * pBody = ( struct vmasd_broadcast_message_body * )( szTempBuffer + sizeof( struct vmasd_broadcast_message_header ) ); // sizeof 값은 352 Byte (64Bit OS )


	// 본 Thread 는 main 에서 vmasd 정보를 수신한 후 동작해야 하므로...  5 sec 정도 기다린 후 동작시킨다.
	sleep( RECV_START_DELAY_SEC );


	// 루프를 돌면서... Queue 에 저장된 vmasd 수신 정보를 POP -> 파싱하여.. 저장 처리한다.
	// g_pStatDataQueue 값은 프로세스 종료 처리시 NULL 로 변경된다.. 따라서 본 Thread 가 종료될 수 있도록 처리한다.
	while( g_pStatDataQueue != NULL )
	{
		// Queue 에서 Data 를 Pop 한다..
		// Queue 객체 자체에서 인자값을 전달한 버퍼를 memset 하므로.. 따로 수행하지 않는다.
		pop_result = g_pStatDataQueue->Pop( szTempBuffer, MAX_SOCKET_BUFFER_SIZE );

		if( pop_result == 0 ) // Pop 성공시..
		{		
			/*
			_LOG( LDBG, "[RecvVmasd:%d] SaveThread: header [%s][0x%04x][%s][%s][%s][%d][%d]", getpid()
				  , pHeader->szTransactionID, ntohl( pHeader->ulMode ), pHeader->szIssuer, pHeader->szServiceID, pHeader->szRCID
				  , ntohl( pHeader->ulResult ), ntohl( pHeader->ulBodyLength ) );		
						
			_LOG( LDBG, "[RecvVmasd:%d] SaveThread: body [%s] [%lu][%lu] [%d][%d] [%d][%d] [%d][%d] [%u][%u][%u][%u] [%s] [%lu]" , getpid()
				  , pBody->domain
				  , pBody->total, pBody->free
				  , pBody->manual_disable, pBody->upload_level
				  , pBody->disk_backlog, pBody->disk_percent
				  , pBody->connection_established, pBody->connection_total
				  , pBody->traffic_ip, pBody->traffic_ib, pBody->traffic_op, pBody->traffic_ob, pBody->hostname
				  , pBody->large_block );
			*/
			
			
			// RC ID 검사는 앞의 수신 처리부에서 했으므로..... pass
			// body 부분 중 domain 항목은 client ip 정보로.. 수신 모듈에서 대체처리 했음.

			// 수신된 정보를 공유메모리에 저장 처리한다.
			if( g_pShmManager->SetVmasdData( pBody ) == false )
			{
				LOG( LERR, "[RecvVmasd:%d] SaveThread: shm update fail.[%s][%s]", getpid(), pBody->domain, pBody->hostname );
			}

		}
		else if( pop_result == 1 )  // Queue 가 empty 인 경우..
		{
			// pop 할 data 가 없으므로.. 잠시 쉰다.. ( 0.5 sec 정도 )
			//_LOG( LDBG, "RecvVmasd[%d]:SaveThread: queue empty. sleep", getpid());
			struct timespec sleep;
			sleep.tv_sec  = 0;
			sleep.tv_nsec = 500000000;  // 0.5 sec
			nanosleep( &sleep, NULL );

		}
		else	// 나머지 상황은 오류..
		{
			LOG( LERR, "[RecvVmasd:%d] SaveThread: queue pop unknown error.[%d]", getpid(), pop_result );
		}
	}

	
	// Thread 종료 관련 처리.
	return NULL;
}

// Worker::RecvVmasd 에서 vmasd 정보를 수신처리하기 위한 Thread.
void * WorkerRecvVmasdClientThread( void * arg )
{
	// 메인 함수에서 이 스레드가 종료되기를 기다리지 않게 분리
	pthread_detach( pthread_self() );

	_LOG( LINF, "[RecvVmasd:%d] ReceiveThread: start...", getpid() );

	// vmasd broadcast data 수신을 위한 준비
	struct sockaddr_in clientSocketAddr;
	socklen_t clientSocketAddrLen = sizeof( clientSocketAddr );

	char szClientAddress[INET_ADDRSTRLEN];
	memset( szClientAddress, 0x00, sizeof( szClientAddress ) );

	char szSocketBuffer[MAX_SOCKET_BUFFER_SIZE];
	int nRead;
	int nResult;

	struct vmasd_broadcast_message_header * pHeader = NULL;
	struct vmasd_broadcast_message_body   * pBody   = NULL;
	
	// vmasd broadcast 정보 수신 대기
	while( 1 )
	{
		// vmasd 로부터 수신된 Data read 대기
		nRead = recvfrom( g_sockVmasdRecv, &szSocketBuffer, MAX_SOCKET_BUFFER_SIZE, 0, ( struct sockaddr * )&clientSocketAddr, &clientSocketAddrLen );

		// 수신 오류 발생시
		if( nRead < 0 )
		{
			LOG( LWAR, "RecvVmasd socket recvfrom error[%d][%s]", errno, strerror( errno ) );
			continue;
		}

		if( inet_ntop( AF_INET, &( clientSocketAddr.sin_addr ), szClientAddress, sizeof( szClientAddress ) ) == NULL )
		{
			LOG( LERR, "RecvVmasd inet_ntop error[%d][%s]", errno, strerror( errno ) );
		}

		// 로그 기록 목적으로 로깅처리
		_LOG( LDBG, "VmasdReceiver[%d][%lu]: vmasd[%s] received. length[%d]", getpid(), pthread_self(), szClientAddress, nRead );

		// 1. vmasd message 의 경우 broadcasting message 이므로.. 사설 네트워크 구간이 분리되지 않은 경우... 다른 RC 의 메시지가 수신될 확률이 높다.
		// 따라서. 동일 RC 에서 전달된 메시지인지 먼저 확인한다.
		pHeader = ( struct vmasd_broadcast_message_header * )szSocketBuffer;
		if( strcasecmp( pHeader->szRCID, CProcessConfig::GetInstance()->GetRcId() ) != 0 )
		{
			// 다른 RC 로 부터 vmasd 메시지를 수신한 경우... 수신 Data 는 그냥 버린다.
			LOG( LWAR, "RecvVmasd: Not match RCID[%s][%s]", pHeader->szRCID, szClientAddress );
			continue;
		}

		// Debug 목적으로 Data 파싱 처리 후 출력...
		/*
		_LOG( LDBG, "BroadcastReceiver[%d]: vmasd [%s][0x%04x][%s][%s][%s][%d][%d]", getpid()
			, pHeader->szTransactionID, ntohl(pHeader->ulMode), pHeader->szIssuer, pHeader->szServiceID, pHeader->szRCID
			, ntohl(pHeader->ulResult), ntohl(pHeader->ulBodyLength) );
		*/
		

		// 2. 기존 rc_vrcd 소스 코드를 참조하여...
		// 수신된 메시시 값 중 ntohl(stHeader->ulMode) == MSG_MODE_VMA_APACHE_DISKSIZE ( 0x0100 ) 인 경우에만 처리..
		// 그외 메시지에 대해서는 Invalid Protocol 로 처리한다.
		if( ntohl( pHeader->ulMode ) != MSG_MODE_VMA_APACHE_DISKSIZE )   // #define MSG_MODE_VMA_APACHE_DISKSIZE  0x0100
		{
			LOG( LWAR, "RecvVmasd: Invalid protocol received.[0x%04x][%s]", ntohl( pHeader->ulMode ), szClientAddress );
		}
		else if( nRead != ( sizeof( struct vmasd_broadcast_message_header ) + sizeof( struct vmasd_broadcast_message_body ) ) )
		{
			// 수신된 packet 크기는 64Bit OS 에서 992 Byte 이어야 함 
			// -  sizeof( struct vmasd_broadcast_message_header ) => 352 (64Bit OS)
			// -  sizeof( struct vmasd_broadcast_message_body )   => 640 (64Bit OS)

			LOG( LWAR, "RecvVmasd: Invalid packet length [%s]. receive[%d]. normal[%lu] "
				 , szClientAddress, nRead
				 , ( sizeof( struct vmasd_broadcast_message_header ) + sizeof( struct vmasd_broadcast_message_body ) ) );
			
		}
		else
		{
			// Data 를 Queue 에 저장하기 전에.. IP 정보를 전달해야 하지만... 저장할 공간이 없으므로...
			// 수신된 body 중 domain, hostname 항목이 중복되지만... domain 항목은 사용하지 않는 항목이므로....
			// domain 항목에 수신된 IP 정보를 기록처리한다.
			pBody = ( struct vmasd_broadcast_message_body * )( szSocketBuffer + sizeof( struct vmasd_broadcast_message_header ) ); // sizeof 값은 352 Byte (64Bit OS )
			memcpy( pBody->domain, szClientAddress, sizeof( szClientAddress ) );
			
			// Data 를 Queue 에 저장.
			if( g_pStatDataQueue != NULL )
			{
				nResult = g_pStatDataQueue->Push( szSocketBuffer, nRead );

				// Push 처리시 반환값
				// -1 : 전달 인자값에 NULL 존재
				// -2 : Queue 생성시 설정한 Data 크기보다 더 큰 nRead 값을 전달한 경우.
				//  1 : Queue 생성시 설정한 최대 Queue 갯수를 초과한 경우..
				//  0 : 정상 처리
				if( nResult != 0 )
				{
					LOG( LWAR, "RecvVmasd: queue push error[%d]", nResult );
				}
			}
		}

	} // while(1)


	// Thread 종료 관련 처리.
	return NULL;
}



// Worker::RecvVmasd 프로세스의 Main 역활을 수행하기 위한 함수.
int WorkerRecvVmasdMain()
{
	// 전역 변수로 설정된 socket 중 해당 worker 에서 사용하지 않는 socket 은 close 처리한다.	
	if( g_sockSystemRecv != -1 )
	{
		close( g_sockSystemRecv );
		g_sockSystemRecv = -1;
	}
	if( g_sockCheckRecv != -1 )
	{
		close( g_sockCheckRecv );
		g_sockCheckRecv = -1;
	}
	if( g_sockInfoSend != -1 )
	{
		close( g_sockInfoSend );
		g_sockInfoSend = -1;
	}


	// Process Rename..
	set_ps_display( PROG_NAME": RecvVmasd", false );
	
	// Set Signal Handler
	SetWorkerRecvVmasdSignalHandler();


	// 사용할 객체 초기화 처리
	// Data 전달을 담당할 Queue 객체 생성
	g_pStatDataQueue = new CFixedQueue( MAX_SOCKET_BUFFER_SIZE, MAX_DATA_QUEUE_COUNT );
	if( g_pStatDataQueue == NULL )
	{
		// 객체 생성 실패시
		LOG( LERR, "[RecvVmasd:%d] data queue create failed. process exit", getpid() );

		ReadyToExitWorkerRecvVmasd();
		return EXIT_FAILURE;
	}

	// Queue 로 전달된 Data 를 파싱하여 공유메모리에 저장 처리를 담당할 Thread 를 생성한다.
	pthread_t workerSaveThreadId;
	if( pthread_create( &workerSaveThreadId, NULL, WorkerRecvVmasdSaveThread, NULL ) != 0 )
	{
		int errorNum = errno;

		LOG( LERR, "[RecvVmasd:%d] save thread create failed[%d][%s]. process exit", getpid(), errorNum, strerror( errorNum ) );

		ReadyToExitWorkerRecvVmasd();
		return EXIT_FAILURE;
	}

	// csagentd 수신 처리 모듈이 먼저 기동되어야.. 공유메모리 동시 write 문제를 회피할 수 있다.
	sleep( RECV_START_DELAY_SEC );
	
	// vmasd 로 부터 UDP broadcasting 정보를 수신 처리할 Thread 생성 처리.
	pthread_t workerReceiveThreadId;
	for( int i = 0; i < MAX_THREAD_RECEIVER; i++ )
	{
		if( pthread_create( &workerReceiveThreadId, NULL, WorkerRecvVmasdClientThread, NULL ) != 0 )
		{
			int errorNum = errno;

			LOG( LERR, "[RecvVmasd:%d] receive thread create failed[%d][%s]. process exit", getpid(), errorNum, strerror( errorNum ) );

			ReadyToExitWorkerRecvVmasd();
			return EXIT_FAILURE;
		}
	}

	sleep( 1 );
	_LOG( LINF, "[RecvVmasd:%d] receive thread count[%d] create success.", getpid(), MAX_THREAD_RECEIVER );
		
		
	// 대기
	while( 1 )
	{
		// 할일 없어서.. 시그널 수신까지 대기 한다.
		pause();
	}
	

	// 종료시 정리 작업 수행
	ReadyToExitWorkerRecvVmasd();
	return EXIT_SUCCESS;
}

