/**
*	@brief 		Send the stream data to the MB86-LSI and recive the stream data from the MB86-LSI.
*	@since		2009/01/08
*	@note		None
*	@attention	None
*	<B><I>COPYRIGHT FUJITSU LIMITED 2009</I></B>
*/

/*
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 */

#include <pthread.h>
#include <unistd.h>
#include <sys/select.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
#include <sys/ioctl.h>
#include <sys/stat.h>
#include <sys/types.h>

#include "FmbCmn.h"

static void fmbStreamClose(enum FMBStremThreadEnum threadKind, struct StreamThread *p_StreamThread);
static void fmbStreamEnd(enum FMBStremThreadEnum threadKind, struct StreamThread *p_StreamThread);
inline BOOL fmbChkPacketSize(unsigned char* stream_buf , int buf_size , int packet_size , int* offset);

static void *fmbRecvStreamThread(void *arg);
static void *fmbSendStreamThread(void *arg);

/**
*	@brief		Start to the thread of stream.
*	@return		TRUE
*	@return		FALSE
*	@note		None
*	@attention	None
*/
BOOL FmbStreamBegin( enum FMBStremThreadEnum	threadKind, HANDLE sceHandle )
{
	int		rc;
	pthread_attr_t threadAtt;
	struct  sched_param threadParam;
	pthread_t	thread;
	int max_priority, min_priority, priority;
	struct StreamThread *p_StreamThread;
	struct FMBComProperty *p_ComProperty;
	struct FMBDeviceInfo *p_deviceInfo;

	FMB_FUNC(TRUE);

	p_deviceInfo = FmbGetDevInfo(SceGetDevNum(sceHandle));
	p_StreamThread = &p_deviceInfo->t_StreamThread_A[threadKind];
	p_ComProperty = FmbGetComProperty(SceGetDevNum(sceHandle));
	if (p_StreamThread->fd < 0){
		FMB_FUNC(FALSE);
		return FALSE;
	}

	if ( p_StreamThread->isThread == TRUE){
		FMB_FUNC(FALSE);
		return FALSE;
	}

	p_StreamThread->fmbfd = open(p_ComProperty->devicePath , O_RDWR);
	ASSERT_SYS(p_StreamThread->fmbfd >= 0, "open()");

	p_StreamThread->fBuf = malloc(p_StreamThread->fSize);
	if ( p_StreamThread->fBuf == NULL ){
		fmbStreamClose(threadKind, p_StreamThread);
		FMB_FUNC(FALSE);
		return FALSE;
	}

	p_StreamThread->isThread = FALSE;
	p_StreamThread->isStop = FALSE;
	p_StreamThread->isForceStop = FALSE;
	p_StreamThread->streamSize = 0;
	p_StreamThread->isFrameUpdate = FALSE;
	p_StreamThread->isSeqEnd = FALSE;
	p_StreamThread->isSystemCmdAck	= FALSE;

	rc = pthread_attr_init(&threadAtt);
	ASSERT_USR(rc == 0, "pthread_attr_init");

	rc = pthread_attr_getschedparam (&threadAtt, &threadParam);
	ASSERT_USR(rc == 0, "pthread_attr_getschedparam");

	max_priority = sched_get_priority_max(SCHED_OTHER);
	min_priority = sched_get_priority_min(SCHED_OTHER);
	priority = p_StreamThread->priority;

	if (priority < min_priority) {
		priority = min_priority;
	}
	else if (max_priority < priority) {
		priority = max_priority;
	}

	threadParam.__sched_priority = priority;

	rc = pthread_attr_setschedparam(&threadAtt, &threadParam);
	ASSERT_USR(rc == 0, "pthread_attr_setschedparam");

	rc = pthread_attr_setdetachstate(&threadAtt , PTHREAD_CREATE_DETACHED);
	ASSERT_USR(rc == 0, "pthread_attr_setdetachstate");

	if (threadKind == FMBEnmStreamRcv) {
		p_StreamThread->start_routine	= fmbRecvStreamThread;
	}
	else {
		p_StreamThread->start_routine	= fmbSendStreamThread;
	}

	rc = pthread_create(	&thread,
				&threadAtt,
				p_StreamThread->start_routine,
				p_StreamThread);
	ASSERT_USR(rc == 0, "pthread_create");
	pthread_attr_destroy(&threadAtt);

	FMB_FUNC(FALSE);
	return TRUE;
}

/**
*	@brief		Thread that receives stream from the MB86 LSI.
*	@param[in]	arg		Ignored
*	@return		NULL
*	@note		None
*	@attention	None
*/
static void *fmbRecvStreamThread(void *arg)
{
	int				r_len;
	int				w_len;
	int				totalLen;
	int 			ps_size=0;
	int 			stream_size=0;
	char*			err_cause;
	BOOL			stream_size_wait=TRUE;
	int				pChk_offset=0;
	BOOL			pChk;
	int				pChkCnt=0;
	BOOL			filewrite=TRUE;
	int				pre_stream_size=0;
	int				diff_stream_size=0;
	int				diff_last_size=0;
	__off64_t       truncate_length;
	enum FMBDeviceNumEnum	deviceNum;
	struct StreamThread *p_StreamThread;
	struct FMBComProperty *p_ComProperty;

	p_StreamThread = (struct StreamThread *) arg;

	InitFuncTraceRecvStream(p_StreamThread->deviceNum);

	FMB_FUNC(TRUE);

	p_StreamThread->isThread = TRUE;
	deviceNum = p_StreamThread->deviceNum;
	p_ComProperty = FmbGetComProperty(deviceNum);

	LogFmb("fmbfd\t%d", p_StreamThread->fmbfd);

	ftruncate64(p_StreamThread->fd,0);
	lseek64(p_StreamThread->fd,0,SEEK_SET);

	if ( p_ComProperty->tsFormat[FMBEnmOutputTsPacket] == FMBEnmTsFormat188 ){
		ps_size = 188;
	}
	else{
		ps_size = 192;
	}

	totalLen = 0;

	while(TRUE){
		if ( (p_StreamThread->isForceStop ==TRUE ) 			||

			 (p_StreamThread->isStop ==TRUE &&
				stream_size_wait == TRUE &&
				p_StreamThread->streamSize <= stream_size) 	||

			 (p_StreamThread->isStop ==TRUE &&
				stream_size_wait == FALSE) 										) {

			LogFmb("[fmbRecvStreamThread]Stream(from LSI) read end..........");
			LogFmb("[fmbRecvStreamThread]Stream Size = %d",stream_size);

			if (stream_size_wait == FALSE ) {
				LogFmb("[fmbRecvStreamThread]Stream size no wait.........");
				break;
			}

			r_len = read(p_StreamThread->fmbfd,
					p_StreamThread->fBuf,
					0);
			if (r_len < 0) {
				if (p_StreamThread->isForceStop ==TRUE) {
					LogFmb("[fmbRecvStreamThread]Stream read error skip (Force Stop).........");
					break;
				}
				fprintOut(stdout, "\x1b[31mERROR: [dev-%d]Failed in receiving the stream.\x1b[m\n", deviceNum);
				LogFmb("ERROR: Failed in receiving the stream.");
			}
			//end..
			break;
		}
		if (stream_size_wait == FALSE ) {
			usleep(300000);
			continue;
		}

		r_len = read(p_StreamThread->fmbfd,
				p_StreamThread->fBuf,
				p_StreamThread->fSize);
		if (r_len < 0) {
			if (p_StreamThread->isForceStop ==TRUE) {
				LogFmb("[fmbRecvStreamThread]Stream read error skip (Force Stop).........");
				continue;
			}
			fprintOut(stdout, "\x1b[31mERROR: [dev-%d]Failed in receiving the stream.\x1b[m\n", deviceNum);
			LogFmb("ERROR: Failed in receiving the stream.");
			stream_size_wait = FALSE;
			continue;
		}

		pre_stream_size = stream_size;
		totalLen += r_len;
		stream_size += totalLen/ps_size;
		totalLen = totalLen%ps_size;

		if( filewrite ){
			if( (p_StreamThread->streamSize !=0 ) &&
					(p_StreamThread->streamSize < stream_size) ){
				LogFmb("[fmbRecvStreamThread]pre_stream_size = %d",pre_stream_size);
				LogFmb("[fmbRecvStreamThread]r_len = %d",r_len);
				diff_stream_size = stream_size - pre_stream_size;
				diff_last_size = stream_size - p_StreamThread->streamSize ;
				r_len = (diff_stream_size - diff_last_size) * ps_size;
				LogFmb("[fmbRecvStreamThread]diff_stream_size = %d",diff_stream_size);
				LogFmb("[fmbRecvStreamThread]diff_last_size = %d",diff_last_size);
				LogFmb("[fmbRecvStreamThread]stream_size = %d",stream_size);
				LogFmb("[fmbRecvStreamThread]Stream Size = %d",p_StreamThread->streamSize);
				LogFmb("[fmbRecvStreamThread]r_len = %d",r_len);

				filewrite=FALSE;
			}

			if ( r_len > 0 ){
				pChk = fmbChkPacketSize(p_StreamThread->fBuf,
										r_len,ps_size,&pChk_offset);
				if( pChk == FALSE ){
					pChkCnt++;
					fprintOut(stdout, "\x1b[31mERROR(dev-%d): Output packet size unmatch occurred.(the %dth packet) Please check settings for packet size and output file.\x1b[m\n" , deviceNum, (stream_size + ( (r_len - pChk_offset)/ps_size )) );
					LogFmb("ERROR: Output packet size unmatch occurred.(the %dth packet) Please check settings for packet size and output file." , (stream_size + ( (r_len - pChk_offset)/ps_size )));
					if(g_dbg.isTS_PSIZE_Exit){
						ASSERT_USR(FALSE, "Output packet size unmatch occurred.");
					}
					while (pChk_offset > ps_size){
						pChk = fmbChkPacketSize(p_StreamThread->fBuf,
												r_len,ps_size,&pChk_offset);
						if( pChk == FALSE ){
							pChkCnt++;
							fprintOut(stdout, "\x1b[31mERROR(dev-%d): Output packet size unmatch occurred.(the %dth packet) Please check settings for packet size and output file.\x1b[m\n" , deviceNum, (stream_size + ( (r_len - pChk_offset)/ps_size )) );
							LogFmb("ERROR: Output packet size unmatch occurred.(the %dth packet) Please check settings for packet size and output file." , (stream_size + ( (r_len - pChk_offset)/ps_size )));
						}
					}
				}
				ASSERT_USR(pChkCnt < FMB_STREAM_PACKET_CHK_TH, "Too many packet size unmatch occurred.");

				w_len = write(p_StreamThread->fd,
						p_StreamThread->fBuf,
						r_len);
				if (w_len < 0) {
					err_cause = strerror(errno);
					LogFmb("write error(%d):%s",errno, err_cause);
					ASSERT_USR(FALSE, "write");
					break;
				}

			}else{
				truncate_length = (__off64_t)p_StreamThread->streamSize * (__off64_t)ps_size;
				LogFmb("[fmbRecvStreamThread] truncate_length = %llu",truncate_length);
				w_len = ftruncate(p_StreamThread->fd,
						truncate_length);
				if (w_len < 0) {
					err_cause = strerror(errno);
					LogFmb("ftruncate error(%d):%s",errno, err_cause);
					ASSERT_USR(FALSE, "ftruncate");
					break;
				}
			}
		}
	}

	fmbStreamEnd(FMBEnmStreamRcv, p_StreamThread);
	FMB_FUNC(FALSE);
	return NULL;
}

/**
*	@brief		Thread that sends stream from the MB86 LSI.
*	@param[in]	arg		Ignored
*	@return		NULL
*	@note		None
*	@attention	None
*/
static void *fmbSendStreamThread(void *arg)
{
	int				r_len;
	int				w_len;
	int				totalLen;
	int 			ps_size=0;
	int 			stream_size=0;
	char*			err_cause;
	int				pChk_offset=0;
	BOOL			pChk;
	int				pChkCnt=0;
	BOOL 			normal_write = TRUE;
	int				endstreamLen;
	enum FMBDeviceNumEnum	deviceNum;
	struct StreamThread *p_StreamThread;
	struct FMBComProperty *p_ComProperty;

	p_StreamThread = (struct StreamThread *) arg;

	InitFuncTraceSendStream(p_StreamThread->deviceNum);

	FMB_FUNC(TRUE);

	p_StreamThread->isThread = TRUE;
	deviceNum = p_StreamThread->deviceNum;
	p_ComProperty = FmbGetComProperty(deviceNum);

	LogFmb("fmbfd\t%d", p_StreamThread->fmbfd);

	if ( p_ComProperty->tsFormat[FMBEnmInputTsPacket] == FMBEnmTsFormat188 ){
		ps_size = 188;
	}
	else{
		ps_size = 192;
	}

	totalLen = 0;
	endstreamLen =0;

	//start_position
	if (p_StreamThread->fileStartPos!=0){
		lseek64(p_StreamThread->fd,
				p_StreamThread->fileStartPos,
				SEEK_SET);
	}
	//end_position
	if (p_StreamThread->fileEndPos!=0){
		endstreamLen = p_StreamThread->fileEndPos /ps_size;
	}

	while(TRUE){

		if ( (p_StreamThread->isForceStop ==TRUE ) ||
		     (p_StreamThread->isStop ==TRUE ) ) {

			LogFmb("[fmbSendStreamThread]Stream(from LSI) write end..........");
			LogFmb("[fmbSendStreamThread]Stream Size = %d",stream_size);

			if ( normal_write == FALSE ) {
				LogFmb("[fmbSendStreamThread] write ng end.........");
				break;
			}

			w_len = write(p_StreamThread->fmbfd,
					p_StreamThread->fBuf,
					0);
			if (w_len < 0) {
				if( p_StreamThread->isForceStop ==TRUE ){
					LogFmb("[fmbSendStreamThread]Stream write error skip (Force Stop).........");
					break;
				}
				switch (p_ComProperty->currentRequest){
					case FMBEnmReqStop :
						LogFmb("[fmbSendStreamThread]Stream write error skip (Stop).........");
						break;
					default:
						ASSERT_USR(FALSE, "write0");
						break;
				}
				break;
			}

			//end..
			break;
		}

		if (normal_write == FALSE ) {
			usleep(300000);
			continue;
		}

		r_len = read(p_StreamThread->fd,
				p_StreamThread->fBuf,
				p_StreamThread->fSize);
		if (r_len < 0) {
			err_cause = strerror(errno);
			LogFmb("read error(%d):%s",errno, err_cause);
			ASSERT_USR(FALSE, "read");
			break;
		}
		if (r_len == 0) {
			LogFmb("[fmbSendStreamThread]File read end..........");
			FmbSetStreamStop(FMBEnmStreamSend, p_StreamThread);
			continue;
		}


		w_len = write(p_StreamThread->fmbfd,
				p_StreamThread->fBuf,
				r_len);

		if (w_len < 0) {
			if( p_StreamThread->isForceStop ==TRUE ){
				LogFmb("[fmbSendStreamThread]Stream write error skip (Force Stop).........");
				continue;
			}
			switch (p_ComProperty->currentRequest){
				case FMBEnmReqStop :
					LogFmb("[fmbSendStreamThread]Stream write error skip (Stop).........");
					continue;
					break;
				default:
					break;
			}
			fprintOut(stdout, "\x1b[31mERROR: Failed in sending the stream.\x1b[m\n");
			LogFmb("ERROR: Failed in sending the stream.");
			normal_write = FALSE;
			continue;
		}

		pChk = fmbChkPacketSize(p_StreamThread->fBuf,
									w_len,ps_size,&pChk_offset);
		if( pChk == FALSE ){
			pChkCnt++;
			fprintOut(stdout, "\x1b[31mERROR: Input packet size unmatch occurred.(the %dth packet) Please check settings for packet size and input file.\x1b[m\n" , (stream_size + ( (w_len - pChk_offset)/ps_size )));
			LogFmb("ERROR: Input packet size unmatch occurred.(the %dth packet) Please check settings for packet size and input file." , (stream_size + ( (w_len - pChk_offset)/ps_size )));
			if(g_dbg.isTS_PSIZE_Exit){
				ASSERT_USR(FALSE, "Input packet size unmatch occurred.");
			}
			while (pChk_offset > ps_size){
				pChk = fmbChkPacketSize(p_StreamThread->fBuf,
											w_len,ps_size,&pChk_offset);
				if( pChk == FALSE ){
					pChkCnt++;
					fprintOut(stdout, "\x1b[31mERROR: Input packet size unmatch occurred.(the %dth packet) Please check settings for packet size and input file.\x1b[m\n" , (stream_size + ( (w_len - pChk_offset)/ps_size )));
					LogFmb("ERROR: Input packet size unmatch occurred.(the %dth packet) Please check settings for packet size and input file." , (stream_size + ( (w_len - pChk_offset)/ps_size )));
				}
			}
		}
		ASSERT_USR(pChkCnt < FMB_STREAM_PACKET_CHK_TH, "Too many packet size unmatch occurred.");

		totalLen += r_len;
		stream_size += totalLen/ps_size;

		switch (p_ComProperty->setMode){
			case FMBEnmFuncModeDec :
				if ( (p_ComProperty->isInternalDecodeStatus != TRUE) &&
					 (p_StreamThread->isFrameUpdate == TRUE )			) {
						LogFmb("[fmbSendStreamThread]Dec 1pic end..........");
						FmbSetStreamStop(FMBEnmStreamSend, p_StreamThread);
						continue;
					}
				break;
			default:
				break;
		}
		totalLen = totalLen%ps_size;
		if ( (endstreamLen != 0) && (stream_size > endstreamLen) ) {
			LogFmb("[fmbSendStreamThread]Reached fileEndPos ..........");
			FmbSetStreamStop(FMBEnmStreamSend, p_StreamThread);
			continue;
		}

	}


	while(TRUE){
		if (p_ComProperty->currentRequest == FMBEnmReqPause ||
				p_ComProperty->currentRequest == FMBEnmReqStartDecFrameAdvance){
			LogFmb("[fmbSendStreamThread] currentRequest=%d",p_ComProperty->currentRequest);
			usleep(300000);
			continue;
		}
		break;
	}

	fmbStreamEnd(FMBEnmStreamSend, p_StreamThread);
	FMB_FUNC(FALSE);

	pthread_exit(EXIT_SUCCESS);
}

/**
*	@brief		ReSet the stream.
*	@param      hScet   Scenario handle
*	@return		None
*	@note		None
*	@attention	None
*/
void FmbStreamRest( HANDLE hScet )
{
	struct FMBDeviceInfo *p_deviceInfo;

	FMB_FUNC(TRUE);


	p_deviceInfo = FmbGetDevInfo(SceGetDevNum(hScet));

	if (p_deviceInfo->t_StreamThread_A[FMBEnmStreamRcv].isThread == TRUE ){
		p_deviceInfo->t_StreamThread_A[FMBEnmStreamRcv].isForceStop = TRUE;
		DrvSetInternal(hScet, FMB_DATA_TYPE_RDMAFLAG , FMB_DATA_DMAFLAG_ON , 0);
		LogFmb("[FmbStreamRest] FMBEnmStreamRcv isForceStop is true.");
	}

	if (p_deviceInfo->t_StreamThread_A[FMBEnmStreamSend].isThread == TRUE ){
		p_deviceInfo->t_StreamThread_A[FMBEnmStreamSend].isForceStop = TRUE;
		DrvSetInternal(hScet, FMB_DATA_TYPE_WDMAFLAG , FMB_DATA_DMAFLAG_ON , 0);
		LogFmb("[FmbStreamRest] FMBEnmStreamSend isForceStop is true.");
	}


	FMB_FUNC(FALSE);
	return;
}

/**
*	@brief		Set the stop flag for recive stream.
*	@param      None
*	@return		None
*	@note		None
*	@attention	None
*/
void FmbSetStreamStop(enum FMBStremThreadEnum	threadKind, struct StreamThread *p_StreamThread)
{
	FMB_FUNC(TRUE);

	if ( threadKind >= FMBEnmStreamMAX ){
		FMB_FUNC(FALSE);
		return;
	}

	if (p_StreamThread->isThread == TRUE ){
		if ( p_StreamThread->isStop != TRUE){
			p_StreamThread->isStop = TRUE;
			LogFmb("[FmbSetStreamStop] id=%d isStop is true.",threadKind);
		}else{
			p_StreamThread->isForceStop = TRUE;
			LogFmb("[FmbSetStreamStop] id=%d isForceStop is true.",threadKind);
		}
	}

	FMB_FUNC(FALSE);
	return;
}

/**
*	@brief		Set the stream size.
*	@param[in]	strm_size	stream size.
*	@return		None
*	@note		None
*	@attention	None
*/
void FmbSetStreamSize(enum FMBStremThreadEnum threadKind,int strm_size, struct StreamThread *p_StreamThread)
{

	FMB_FUNC(TRUE);

	if ( threadKind != FMBEnmStreamRcv ){
		FMB_FUNC(FALSE);
		return;
	}

	if (p_StreamThread->isThread == TRUE ){
		p_StreamThread->streamSize = strm_size;
		LogFmb("[FmbSetStreamSize] id=%d Stream Size =%d",threadKind,p_StreamThread->streamSize);
		if ( strm_size == 0 ){
			p_StreamThread->isStop = TRUE;
			LogFmb("[FmbSetStreamSize] id=%d isStop is true.",threadKind);
		}
	}

	FMB_FUNC(FALSE);
	return;
}
/**
*	@brief		Set the frame update flag.
*	@param[in]	None
*	@return		None
*	@note		None
*	@attention	None
*/
void FmbSetStreamFrameUpdate( enum FMBStremThreadEnum threadKind, struct StreamThread *p_StreamThread )
{
	FMB_FUNC(TRUE);

	if ( threadKind != FMBEnmStreamSend ){
		FMB_FUNC(FALSE);
		return;
	}

	if (p_StreamThread->isThread == TRUE ){
		p_StreamThread->isFrameUpdate = TRUE;
		LogFmb("[FmbSetStreamFrameUpdate] id=%d FrameUpdate is true.",threadKind);
	}


	FMB_FUNC(FALSE);
	return;
}
/**
*	@brief		Set the sequence end flag.
*	@param[in]	None
*	@return		None
*	@note		None
*	@attention	None
*/
void FmbSetStreamSeqEnd( enum FMBStremThreadEnum threadKind, struct StreamThread *p_StreamThread )
{
	FMB_FUNC(TRUE);

	if ( threadKind != FMBEnmStreamSend ){
		FMB_FUNC(FALSE);
		return;
	}

	if (p_StreamThread->isThread == TRUE ){
		p_StreamThread->isSeqEnd = TRUE;
		LogFmb("[FmbSetStreamSeqEnd] id=%d Sequence End is true.",threadKind);
	}

	FMB_FUNC(FALSE);
	return;
}
/**
*	@brief		Set the sequence end flag.
*	@param[in]	None
*	@return		None
*	@note		None
*	@attention	None
*/
void FmbSetSystemCmdAck( enum FMBStremThreadEnum threadKind, struct StreamThread *p_StreamThread )
{
	FMB_FUNC(TRUE);

	if ( threadKind != FMBEnmStreamSend ){
		FMB_FUNC(FALSE);
		return;
	}

	if (p_StreamThread->isThread == TRUE ){
		p_StreamThread->isSystemCmdAck = TRUE;
		LogFmb("[FmbSetSystemCmdAck] id=%d System Command Ack.",threadKind);
	}

	FMB_FUNC(FALSE);
	return;
}

/**
*	@brief		Close the stream thread descriptor.
*	@param[in]	None
*	@return		None
*	@note		None
*	@attention	None
*/
static void fmbStreamClose( enum FMBStremThreadEnum threadKind, struct StreamThread *p_StreamThread)
{
	FMB_FUNC(TRUE);

	if ( threadKind >= FMBEnmStreamMAX ){
		return;
	}

	close(p_StreamThread->fmbfd );
	close(p_StreamThread->fd );

	p_StreamThread->fmbfd = -1;
	p_StreamThread->fd = -1;

	LogFmb("[fmbStreamClose] id=%d Close.",threadKind);

	FMB_FUNC(FALSE);

	return;
}
/**
*	@brief		Finished the stream thread.
*	@param[in]	threadKind
*	@param[in]	deviceNum
*	@return		None
*	@note		None
*	@attention	None
*/
static void fmbStreamEnd( enum FMBStremThreadEnum	threadKind, struct StreamThread *p_StreamThread)
{
	struct RecvEventArgType event;

	if ( threadKind >= FMBEnmStreamMAX ){
		return;
	}
	FMB_FUNC(TRUE);

	fmbStreamClose(threadKind, p_StreamThread);
	free(p_StreamThread->fBuf);

	event.devNum = p_StreamThread->deviceNum;
	event.dev = FMBEnmCoreDeviceStream;
	event.evnt = FMBEnmCoreFromSteamEndMsg;
	event.WaitDetail.StreamMsg.id = threadKind;

	p_StreamThread->isThread = FALSE;
	LogFmb("[fmbStreamEnd] id=%d isThread is false.",threadKind);

	FmbRecvSendEvent(&event, *p_StreamThread->p_fdPipeWrite);
	LogFmb("[fmbStreamEnd]FmbRecvSendEvent dev=%d evnt=%d id=%d",event.dev,event.evnt,event.WaitDetail.StreamMsg.id);


	FMB_FUNC(FALSE);


	return;
}
/**
*	@brief		Open the stream file.
*	@param[in]	FmbFileTypeEnum file kind(input file : FmbEnmInFile / output file : FmbEnmOutFile)
*	@param[in]	filepath file path name
*	@return		FmbEnmFileNormal    Normal end
*	@return		FmbEnmFileAccesFail File access fail
*	@return		FMBEnmFilePacketSizeFail File Packet size fail
*	@return		FmbEnmFileOtherFail File Open fail
*	@note		None
*	@attention	None
*/
enum FMBFileResultEnum FmbFileOpen(enum FMBFileEnum fileenum , char* filepath , int* errPacketCnt)
{
	unsigned char*   fBuf;
	int				r_len,chk_len=0,ps_size;
	BOOL			chkPsize;

	FMB_FUNC(TRUE);

	if (gp_FmbDevInfo->t_StreamThread_A[fileenum].isThread == TRUE){
		FMB_FUNC(FALSE);
		return FMBEnmFileExecutionFail;
	}

	switch (fileenum){
		case FMBEnmInFile:
			//FMB_STREAM_THREAD_SEND
			if (gp_FmbDevInfo->t_StreamThread_A[fileenum].fd >= 0){
				close(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd );
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = -1;
			}
			gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = open(filepath,O_RDONLY|O_LARGEFILE);
			break;
		case FMBEnmOutFile:
			//FMB_STREAM_THREAD_RCV
			if (gp_FmbDevInfo->t_StreamThread_A[fileenum].fd >= 0){
				close(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd );
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = -1;
			}
			if(FmbIsExistenceFile(filepath)){
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = open(filepath,O_WRONLY|O_APPEND|O_LARGEFILE,S_IRWXU );
			}else{
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = open(filepath,O_WRONLY|O_CREAT|O_TRUNC|O_LARGEFILE,S_IRWXU );
			}
			break;
		default:
			ASSERT_USR(FALSE, "Invalid file kind.");
			FMB_FUNC(FALSE);
			return FMBEnmFileOtherFail;
	}
	if (gp_FmbDevInfo->t_StreamThread_A[fileenum].fd < 0){
		gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = -1;
		switch (errno) {
		case EACCES :
			FMB_FUNC(FALSE);
			return FMBEnmFileAccesFail;
		default:
			FMB_FUNC(FALSE);
			return FMBEnmFileOtherFail;
		}
	}


	switch (fileenum){
		case FMBEnmInFile:
			fBuf = malloc(FMB_STREAM_THREAD_SEND_FSIZE);
			if ( fBuf == NULL ){
				ASSERT_USR(FALSE, "malloc NG.");
				close(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd );
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = -1;
				FMB_FUNC(FALSE);
				return FMBEnmFileOtherFail;
			}
			r_len = read(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd,
					fBuf,
					FMB_STREAM_THREAD_SEND_FSIZE);
			if (r_len < 0) {
				free(fBuf);
				close(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd );
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = -1;
				ASSERT_USR(FALSE, "read");
				FMB_FUNC(FALSE);
				return FMBEnmFileOtherFail;
			}

			if ( gp_FmbComProperty->tsFormat[FMBEnmInputTsPacket] == FMBEnmTsFormat188 ){
				ps_size = 188;
			}
			else{
				ps_size = 192;
			}

			chkPsize = fmbChkPacketSize(fBuf,r_len,ps_size,&chk_len);
			if( chkPsize == FALSE ){
				*errPacketCnt = (r_len - chk_len)/ps_size ;
				if (*errPacketCnt ==0 ){
					*errPacketCnt = 1;
				}
				free(fBuf);
				close(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd );
				gp_FmbDevInfo->t_StreamThread_A[fileenum].fd = -1;
				FMB_FUNC(FALSE);
				return FMBEnmFilePacketSizeFail;
			}
			lseek(gp_FmbDevInfo->t_StreamThread_A[fileenum].fd,0,SEEK_SET);
			free(fBuf);
			break;
		default:
			break;
	}

	FMB_FUNC(FALSE);
	return FMBEnmFileNormal;
}

/**
*	@brief		Get the stream file position.
*	@param[in]	FMBFilePosEnum file position kind(start : FMBEnmStartPos / end : FMBEnmEndPos)
*	@return		position
*	@note		None
*	@attention	None
*/
__off64_t FmbGetStrmFilePos(enum FMBFilePosEnum filPoseenum )
{
	__off64_t position=0;

	FMB_FUNC(TRUE);

	switch(filPoseenum){
	case FMBEnmStartPos:
		position = gp_FmbDevInfo->t_StreamThread_A[FMBEnmInFile].fileStartPos;
		break;
	case FMBEnmEndPos:
		position = gp_FmbDevInfo->t_StreamThread_A[FMBEnmInFile].fileEndPos;
		break;
	default:
		break;
	}

	FMB_FUNC(FALSE);
	return (position);
}


/**
*	@brief		Set the stream file position.
*	@param[in]	FMBFilePosEnum file position kind(start : FMBEnmStartPos / end : FMBEnmEndPos)
*	@return		position
*	@note		None
*	@attention	None
*/
void FmbSetStrmFilePos(enum FMBFilePosEnum filPoseenum , __off64_t position)
{

	FMB_FUNC(TRUE);

	switch(filPoseenum){
	case FMBEnmStartPos:
		gp_FmbDevInfo->t_StreamThread_A[FMBEnmInFile].fileStartPos = position ;
		break;
	case FMBEnmEndPos:
		gp_FmbDevInfo->t_StreamThread_A[FMBEnmInFile].fileEndPos = position;
		break;
	default:
		break;
	}

	FMB_FUNC(FALSE);
	return;
}


/**
*	@brief		Checked the stream of packet size.
*	@param[in]	stream buffer
*	@param[in]	stream buffer size
*	@param[in]	packet size
*	@param[in]	packet size offset
*	@return		0-  remain size
*	@note		None
*	@attention	None
*/
inline BOOL fmbChkPacketSize( unsigned char* stream_buf , int buf_size , int packet_size , int* offset)
{
	BOOL hitF;
	int remain_size;

	remain_size = buf_size;

	if ( *offset == 0 ){
		hitF = FALSE;
	}
	else if( *offset < packet_size ){
		hitF = TRUE;
		stream_buf+=(packet_size-*offset);
		remain_size-=(packet_size-*offset);
	}
	else{ /* *offset >= packet_size */
		hitF = FALSE;
		stream_buf+=(remain_size-*offset);
		remain_size=*offset;
	}

	while( remain_size >= packet_size ){
		if(*stream_buf==0x47){
			hitF = TRUE;
			stream_buf+=packet_size;
			remain_size-=packet_size;
		}
		else{
			if (hitF == TRUE){
				*offset=remain_size;
				return FALSE ;
			}
			stream_buf++;
			remain_size--;
		}
	}

	*offset=remain_size;
	return hitF ;
}