
#include "MessageSender.h"
#include "hi_ctrl.h"
#include <iconv.h>
#include "file_no.h"

char g_szProcessname[MAX_STRING_SIZE + 1];
char g_szProcessPath[PATH_MAX + 1];

int g_nResponseTimeout;
char szOrganizationID[6+1];
char szWorkType[4+1];
char szMsgType[4+1];
char szRespCode[] = "0000";
//char *szData;
char szData[MAX_STRING_SIZE+1];
char rtnData[MAX_STRING_SIZE+1];

extern int g_nSerailNo = 1;
pthread_mutex_t g_sSerialMutex = PTHREAD_MUTEX_INITIALIZER;

static    struct    {
    int    cmd;                /* CONTROL COMMAND          */
    int    act;                /* 0:송신 1:응답 2:3자통신  */
    int    work;
    char  *cmd_nm;
} web_act[] = {
	{ FNO_BASICTBL       , 9001,   10 , " BASIC TABLE               "},
	{ FNO_WRKRINFO       , 9002,   10 , " WORKER TABLE              "},
    { FNO_BEFMST         ,   21,   10 , " ADV_MASTER TABLE COUNT    "},
	{ FNO_BEFCNG         ,   21,   30 , " ADV_CHANGE TABLE          "},
    { FNO_IMM_CARNO      ,   22,   10 , " EXEMPT_MASTER TABLE COUNT "},
	{ FNO_IMM_CARNO_CHG  ,   22,   30 , " EXEMPT_CHANGE TABLE       "},
    { FNO_IMM_CARDNO     ,   23,   10 , " RFID_MASTER TABLE COUNT   "},
	{ FNO_IMM_CARDNO_CHG ,   23,   30 , " RFID_CHANGE TABLE         "},
    { FNO_RED_CARNO      ,   24,   10 , " RED_MASTER TABLE COUNT    "},
	{ FNO_RED_CARNO_CHG  ,   24,   30 , " RED_CHANGE TABLE          "},
	{ FNO_DAYMNGT        ,   31,   10 , " ADV_DAY_END_CALL          "},
	{ FNO_DAY_RTN        ,   32,   10 , " ADV_DAY_END_RETURN        "},
    { FNO_EGREEN_RTN     ,   33,   10 , " Egreen_Return             "},
//2020.03.03 RFID Web Server Check 추가
    { FNO_WEB_CHK        , 9011,   10 , " Web Server File Check     "},
//2020.03.09 WAS 마스터 파일 생성 Check 추가
    { FNO_MST_CHK        , 9010,   10 , " Web Master File Check     "},
	{ FNO_CHG_CHK        , 9010,   20 , " Web Change File Check     "},
//2020.05.13 개정변경요청 추가
    { FNO_RVS_UPT        , 9020,   10 , " Basic Rvs Update          "},
    { -1                 ,   -1,   -1 , "                           "}
};

int MessageSender(int kind, char *packet, char *path, char *rtn)
{
    char szProcessname[PATH_MAX + 1];
    char *sDataMalloc;
	struct write_info wi;
    int nResult;
    int i = 0;

proc_log("MessageSender Start\n");
proc_log("---------------------------[%s]-----------------------\n", packet);
    WriteLog(LogType_SystemOut,LogLevel_Info, "Process Started![%s]", g_szProcessname);

    memset(szData, 0x00, MAX_STRING_SIZE);
	memset(rtnData, 0x00, MAX_STRING_SIZE);
	sprintf(szData, "%s", packet);
    sprintf(wi.plz_id, "%.2s", PLZ_CODE2);

    while(1)
    {
        if(web_act[i].cmd == kind || web_act[i].cmd == -1) break;
        i++;
    }

    if(web_act[i].act == -1)
    {
        proc_log("지정된 작업이 없습니다[%02d]\n", kind);
        return -1;
    }

    snprintf(szOrganizationID, 6+1, "%06d", lb_data2atoi(wi.plz_id, 2));
    snprintf(szWorkType, 4+1, "%04d", web_act[i].act);
	snprintf(szMsgType, 4+1, "%04d", web_act[i].work);
    //2019.06.27 InitDoubleLinkedList Main 선언
    InitThread_Before();
proc_log("Queue 초기화[%s][%s][%s]\n", szOrganizationID, szWorkType, szMsgType);

    nResult = InitThreadWait(ProcThread);
	//nResult = InitThread(ProcThread);

    snprintf(rtn, strlen(rtnData)+1, "%s", rtnData);
	//SafeFree((void **)&szData);
	//SafeFree((void **)&rtnData);
	//SafeFree((void **)&sDataMalloc);
    //WriteLog(LogType_SystemOut,LogLevel_Info, "Process Stopped![%s]", g_szProcessname);

    return 0;
}

int InitMessageSenderConfig(char *path)
{
    char szConfigfilename[PATH_MAX + 1];
    char szConfigfilenameTemp[PATH_MAX + 1];
    char szValueTemp[MAX_STRING_SIZE + 1];
	char szProcessname[PATH_MAX + 1];
	char *pFindPointer;
    int nResult;

    nResult = SafeStrCopy(szProcessname, sizeof(szProcessname), path);

    if (nResult != SEC_ERR_SUCCESS)
    {
        printf("strcpy_s Error![%d]\n", nResult);
    }

    pFindPointer = szProcessname;
    nResult = SafeStrrChr(szProcessname, strnlen_s(szProcessname, sizeof(szProcessname)), PATH_SEPERATOR, &pFindPointer);

    if (pFindPointer == NULL)
    {
        nResult = SafeStrCopy(g_szProcessname, sizeof(g_szProcessname), szProcessname);
    }
    else
    {
        nResult = SafeStrCopy(g_szProcessname, sizeof(g_szProcessname), pFindPointer + 1);
    }

    GetExecPath(g_szProcessPath, sizeof(g_szProcessPath) - 1, path);

    if (g_szProcessPath == NULL)
    {
        printf("Can't get process path[Cann't load config file]!\nExit Program!");
        return -1;
    }

    snprintf(szConfigfilenameTemp, sizeof(szConfigfilenameTemp), "%s%c%s%c", g_szProcessPath, PATH_SEPERATOR, CONF_FILE_PATH, PATH_SEPERATOR);

    realpath(szConfigfilenameTemp, szConfigfilename);

    snprintf(szValueTemp, sizeof(szValueTemp), "%c", PATH_SEPERATOR);
    nResult = SafeStrCat(szConfigfilename, sizeof(szConfigfilename), szValueTemp, 1);
    nResult = SafeStrCat(szConfigfilename, sizeof(szConfigfilename), g_szProcessname, strnlen_s(g_szProcessname, sizeof(g_szProcessname)));
    nResult = SafeStrCat(szConfigfilename, sizeof(szConfigfilename), CONF_FILE_EXT, strnlen_s(CONF_FILE_EXT, sizeof(CONF_FILE_EXT)));

    printf("Process Name : %s\n", g_szProcessname);
    printf("Process Path : %s\n", g_szProcessPath);
    printf("Config Filename : %s[%d]\n", szConfigfilename, nResult);

    ReadConfigFile(true, szConfigfilename);

    // Read MS_RESPONSE_TIMEOUT_KEY
    if (GetConfigValue(MS_RESPONSE_TIMEOUT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket MS_RESPONSE_TIMEOUT_KEY configuration not found!");
        return SOCK_ERR_CONFIG;
    }

    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nResponseTimeout) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket MS_RESPONSE_TIMEOUT_KEY configuration error[ConfigFile:%s]!", szValueTemp);
        return SOCK_ERR_CONFIG;
    }

    // Read CONNECTION_TYPE
    if (GetConfigValue(SOCK_CONF_CONN_TYPE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket CONNECTION_TYPE configuration not found[Available Value:Server or Client]!");
        return SOCK_ERR_CONFIG;
    }

    if (MappingStringToInt(szValueTemp, sizeof(szValueTemp), g_sConnTypeInfo, sizeof(g_sConnTypeInfo)/sizeof(struct VALUE_STRING_PAIR_INFO), &g_eConnectionType) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket Connection Type configuration error[ConfigFile:%s,Available Value:Server or Client]!", szValueTemp);
        return SOCK_ERR_CONFIG;
    }

    if (InitLogConfig() != LOG_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }

    if (InitSenderMQConfig() != MQ_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }

    if (InitThreadConfig() != THREAD_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }

    return MS_ERR_SUCCESS;
}

void *ProcThread(void *pData)
{
    char cOccurType;
    struct THREAD_INFO *pThreadInfo;
    struct THREAD_INFO SendThreadInfo;
    char szMessageBuffer[MAX_STRING_SIZE + 1];
    struct MESSAGE_INFO sMessageInfo;
    struct MESSAGE_INFO sndMessageInfo;
    int nResult;
    int nRslt = 0;
    int *pnThreadExitCode;
	amqp_queue_purge_ok_t *pQueuePurgeResult;
    char szTopic[MAX_STRING_SIZE + 1];
    char szTargetTopic[MAX_STRING_SIZE + 1];
    char szReplyQueuename[MAX_STRING_SIZE + 1];
    char szReply2Queuename[MAX_STRING_SIZE + 1];
    char szExchangername[MAX_STRING_SIZE + 1];
    uuid_t sUUIDTemp;
    char szSendUUIDString[UUID_STRING_SIZE + 1];
    char szSend2UUIDString[UUID_STRING_SIZE + 1];
    char szRecvUUIDString[UUID_STRING_SIZE + 1];
    char szRecv2UUIDString[UUID_STRING_SIZE + 1];
	
	char procData[MAX_STRING_SIZE + 1]; memset(procData, 0x00, sizeof(procData));
	char webData[MAX_STRING_SIZE + 1]; memset(webData, 0x00, sizeof(webData));
    int nMessageLength;
    //int nPacketLength;
    int nSerialNo;
    int nCompareResult, size, ii;

    //time_t sNow;
    struct tm tNow;
    int nMillisecond;

    //struct timespec sRecvTimeSpec;
    //time_t sRecvNow;
    struct tm tRecvNow;
    int nRecvMillisecond;

    struct timeval sResponseTimeout;

    uint64_t nDeliveryTag;

    int nDiffSec;
    int nDiffMilliSec;
    long nDiffTime;

    bool bMessageMatch;
    bool bTimeout;

    memcpy(procData, szData, sizeof(procData));

    pThreadInfo = (struct THREAD_INFO *)pData;
    SafeMemSet(&SendThreadInfo, sizeof(SendThreadInfo), 0, sizeof(SendThreadInfo));
    pthread_mutex_init(&SendThreadInfo.sContextMutex, NULL);
    pthread_mutex_init(&SendThreadInfo.sPublishMQ.sMQMutex, NULL);
    pthread_mutex_init(&SendThreadInfo.sConsumeMQ.sMQMutex, NULL);
    pnThreadExitCode = NULL;
    nResult = SafeMemoryAlloc((void **)&pnThreadExitCode, sizeof(int));
    SendThreadInfo.nThreadIndex = pThreadInfo->nThreadIndex;
    SendThreadInfo.bThreadLoop = true;
    //pnThreadExitCode = SafeAlloc(pnThreadExitCode, sizeof(int));
    *pnThreadExitCode = 0;
    //nSerialNo = 1;
    if (g_eConnectionType == SOCK_CONN_TYPE_SERVER)
    {
        cOccurType = 'S';
    }
    else
    {
        cOccurType = 'C';
    }
/*
    while (pThreadInfo->bThreadLoop)
    {
*/
        if (pThreadInfo->sPublishMQ.bMQConnection == false || pThreadInfo->sPublishMQ.bMQChannelOpen == false
            || pThreadInfo->sConsumeMQ.bMQConnection == false || pThreadInfo->sConsumeMQ.bMQChannelOpen == false)
        {
            if (ConnectSenderMQ(pThreadInfo, true) != MQ_ERR_SUCCESS)
            {
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Connect SenderMQ is failed!", pThreadInfo->nThreadIndex);
//                continue;
	            pThreadInfo->nRunState = SOCK_THREAD_STATE_TERMINATED;
			    pthread_mutex_destroy(&pThreadInfo->sContextMutex);
				pthread_mutex_destroy(&pThreadInfo->sPublishMQ.sMQMutex);
				pthread_mutex_destroy(&pThreadInfo->sConsumeMQ.sMQMutex);
				SafeFree((void **)&pThreadInfo);
				SafeFree((void **)&pnThreadExitCode);
                return -1;
            }
            
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Connect SenderMQ is succeeded!!", pThreadInfo->nThreadIndex);
            snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendRespQueuename, szOrganizationID, pThreadInfo->nThreadIndex);
        }

        GetCurrTime(&tNow, &nMillisecond);
        nSerialNo = GetSerialNo();
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s%c%s%s%08d%04d%02d%02d%02d%02d%02d%s%s", szOrganizationID, cOccurType, szWorkType,
                 szMsgType, nSerialNo++, tNow.tm_year + 1900, tNow.tm_mon + 1, tNow.tm_mday, tNow.tm_hour, tNow.tm_min, tNow.tm_sec, szRespCode, procData);
        nMessageLength = strnlen_s(szMessageBuffer, sizeof(szMessageBuffer));
        sMessageInfo.nMessageLength = nMessageLength;
        sMessageInfo.paMessageData = NULL;

        SafeMemoryAlloc((void **)&sMessageInfo.paMessageData, sMessageInfo.nMessageLength + 5);

        snprintf(sMessageInfo.paMessageData, sMessageInfo.nMessageLength + 5, "%04X", sMessageInfo.nMessageLength);
        SafeStrCat(sMessageInfo.paMessageData + 4, sMessageInfo.nMessageLength + 1, szMessageBuffer, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)));
        snprintf(szTopic, sizeof(szTopic), "%s.%s%s", szOrganizationID, szWorkType, szMsgType);
#ifdef _WIN32
        uuid_generate(&sUUIDTemp);
        uuid_unparse(&sUUIDTemp, szSendUUIDString);
#else
        uuid_generate(sUUIDTemp);
        uuid_unparse(sUUIDTemp, szSendUUIDString);
#endif

        nResult = PutMessageToQueue(pThreadInfo->sPublishMQ.sMQConnectionState, &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen, pThreadInfo->nThreadIndex,
                                    g_szMQMessageSendExchangername, szReplyQueuename, szTopic, &sMessageInfo, sMessageInfo.paMessageData, sMessageInfo.nMessageLength,
                                    szSendUUIDString, sizeof(szSendUUIDString), false, NULL);
        proc_log("SendData[%d][%s] : [Thread:%02d]%s\n", nResult, szReplyQueuename, pThreadInfo->nThreadIndex, sMessageInfo.paMessageData);
        if (nResult != MS_ERR_SUCCESS)
        {
			proc_log("[Thread:%02d]MQ Data Publish Error[Data:%s]!", pThreadInfo->nThreadIndex, sMessageInfo.paMessageData);
            // 메시지 전송 오류인 경우 처리
            // 오류인 경우 오류 로그에 추가하는 형태의 샘플이지만 DB 처리라면 재전송 여부를 결정해서 재전송 몇번 이내면 재전송 하고 그렇지 않으면 실패 처리하는 로직이 필요하다.
            WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Publish Error[Data:%s]!", pThreadInfo->nThreadIndex, sMessageInfo.paMessageData);
        }
        else
        {
            WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'P', g_szMQMessageSendExchangername, szTopic, szSendUUIDString, sMessageInfo.nMessageLength, "%s",
                          sMessageInfo.paMessageData);
            WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Publish[Data:%s]!", pThreadInfo->nThreadIndex, sMessageInfo.paMessageData);
            bMessageMatch = false;
            bTimeout = false;

            if (g_bBypassMode)
            {
                do
                {
                    GetCurrTime(&tRecvNow, &nRecvMillisecond);
                    nDiffTime = TimeDiff(&tNow, nMillisecond, &tRecvNow, nRecvMillisecond);
                    nDiffSec = nDiffTime / 1000;
                    nDiffMilliSec = nDiffTime % 1000;

                    if (nDiffMilliSec < 0)
                    {
                        nDiffSec--;
                        nDiffMilliSec += 1000;
                    }

                    sResponseTimeout.tv_sec = g_nResponseTimeout / 1000 - nDiffSec;
                    sResponseTimeout.tv_usec = (g_nResponseTimeout % 1000 - nDiffMilliSec) * 1000;
                    if (sResponseTimeout.tv_usec < 0)
                    {
                        sResponseTimeout.tv_sec--;
                        sResponseTimeout.tv_usec += 1000000;
                    }
                    if (sResponseTimeout.tv_sec < 0 || sResponseTimeout.tv_usec < 0)
                    {
                        WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]Response Timeout 1! - UUID:%s, Message:%s, ProcessTime:%d.%03ds\n", pThreadInfo->nThreadIndex, szSendUUIDString,
                                                                      sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                        bTimeout = true;
                        break;
                    }
					memcpy(szRecvUUIDString, szSendUUIDString, sizeof(szRecvUUIDString));
					memset(szMessageBuffer, 0x00, sizeof(szMessageBuffer));
                    nResult = GetMessageFromQueue(pThreadInfo->sConsumeMQ.sMQConnectionState, szExchangername, sizeof(szExchangername), szMessageBuffer, sizeof(szMessageBuffer),
												szRecvUUIDString, sizeof(szRecvUUIDString), szReplyQueuename, sizeof(szReplyQueuename), &pThreadInfo->sConsumeMQ.bMQConnection,
												&pThreadInfo->sConsumeMQ.bMQChannelOpen, &nDeliveryTag, &sResponseTimeout);
proc_log("Get Msg[%s][%s][%s][%d]\n", szMessageBuffer, szRecvUUIDString, szReplyQueuename, nResult);
                    GetCurrTime(&tRecvNow, &nRecvMillisecond);
                    nDiffTime = TimeDiff(&tNow, nMillisecond, &tRecvNow, nRecvMillisecond);
                    nDiffSec = nDiffTime / 1000;
                    nDiffMilliSec = nDiffTime % 1000;

                    if (nDiffMilliSec < 0)
                    {
                        nDiffSec--;
                        nDiffMilliSec += 1000;
                    }

                    if (nResult == MQ_ERR_TIMEOUT)
                    {
                        //printf("Response Timeout! - UUID:%s, Message:%s, ProcessTime:%d.%03ds\n", );
                        WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]Response Timeout 2! - UUID:%s, Message:%s, ProcessTime:%d.%03ds\n",
                                 pThreadInfo->nThreadIndex, szSendUUIDString, sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                        bTimeout = true;
                        break;
                    }

                    nResult = SafeStrCmp(szRecvUUIDString, sizeof(szRecvUUIDString), szSendUUIDString, &nCompareResult);
                    if (nCompareResult != 0)
                    {
                        WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]Recv UUID is not matched! - UUID:[Send:%s, Recv:%s], Message:%s, ProcessTime:%d.%03ds\n",
								pThreadInfo->nThreadIndex, szSendUUIDString, szRecvUUIDString, sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                        amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
                        continue;
                    }
                    bMessageMatch = true;
                    nMessageLength = strnlen_s(szMessageBuffer, sizeof(szMessageBuffer));
                    amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
                    WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'C', szReplyQueuename, "", szRecvUUIDString, nMessageLength, "%s", szMessageBuffer);
                    WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Consume[Data:%s] - ProcessTime:%d.%03ds!", pThreadInfo->nThreadIndex, szMessageBuffer, nDiffSec, nDiffMilliSec);
                } while (bMessageMatch == false && bTimeout == false);
                if (!bTimeout)
                {
					memset(rtnData, 0x00, sizeof(rtnData));
					proc_log("Get Msg[%s][%d]\n", szMessageBuffer, nResult);
					snprintf(rtnData, strnlen_s(szMessageBuffer+45, sizeof(szMessageBuffer)-45)+1, "%s", szMessageBuffer+45);
					proc_log("Rtn Data[%s]\n", rtnData);
                    // 전송 완료 처리 필요
                }
            }
        }
        //sleep(1); // test
/*
    }
*/
    CloseMQ(pThreadInfo->nThreadIndex, pThreadInfo->sPublishMQ.sMQConnectionState, &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen, &pThreadInfo->sPublishMQ.sMQMutex);
    CloseMQ(pThreadInfo->nThreadIndex, pThreadInfo->sConsumeMQ.sMQConnectionState, &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen, &pThreadInfo->sConsumeMQ.sMQMutex);
#if 0 //2019.07.12 메모리 증가로 Mutex Unlock 후 삭제 부분 추가
	pThreadInfo->nRunState = SOCK_THREAD_STATE_TERMINATED;
    pthread_mutex_destroy(&pThreadInfo->sContextMutex);
    pthread_mutex_destroy(&pThreadInfo->sPublishMQ.sMQMutex);
    pthread_mutex_destroy(&pThreadInfo->sConsumeMQ.sMQMutex);
    SafeFree((void **)&pThreadInfo);
    SafeFree((void **)&sMessageInfo.paMessageData);
    SafeFree((void **)&pnThreadExitCode);
#endif
    return 0;
}

int GetSerialNo()
{
    int nSerialNo;
    pthread_mutex_lock(&g_sSerialMutex);
    nSerialNo = g_nSerailNo++;
    pthread_mutex_unlock(&g_sSerialMutex);
    return nSerialNo;
}
