
#include "MessageSender.h"
#include "hi_ctrl.h"
#include "taglnk_frame.h"
#include "tagbl_struc.h"

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

int g_nResponseTimeout;
int cmd;
int type;
int job;

char szOrganizationID[6+1];
char szWorkType[4+1];
char szMsgType[4+1];
//char szSerialID[8 + 1] = "00000001";
//char szRequestDTime[14 + 1] = "20180917014900";
char szRespCodeT[] = "0010";
char szData[MAX_STRING_SIZE + 1];
char rtnData[MAX_STRING_SIZE + 1];

int g_nSerailNoT = 1;
pthread_mutex_t g_sSerialMutexT = PTHREAD_MUTEX_INITIALIZER;

static    struct    {
    int    cmd;                /* CONTROL COMMAND          */
    int    act;                /* 0:송신 1:응답 2:3자통신  */
    int    work;
    char  *cmd_nm;
} lane_act[] = {
    { NS_LANE_STAT          ,   1,    1 , "Web Service Lane State     "},/* 0x75 차로상태(STATUS) */
	{ NS_HND_DATA           ,   1,    2 , "Web Process Data           "},/* 0x17 처리데이터 */
    { -1                    ,  -1,   -1 , "                           "}
};

int MessageSenderHTTP(int kind, char *packet, char *path, char *rtn)
{
    char szProcessname[PATH_MAX + 1];
    char *pFindPointer;
	struct write_info wi;
    struct rcv_lane_data *rd;
    struct bl_struct *sd;
    int nResult;
    int i = 0;

lane_log("--------------------------------------------------\n");
    WriteLog(LogType_SystemOut,LogLevel_Info, "Process Started![%s]", g_szProcessname);

    while(1)
    {
        if(lane_act[i].cmd == kind || lane_act[i].cmd == -1) break;
        i++;
    }
    cmd = lane_act[i].cmd;
lane_log("lane_act[%d].cmd[%02x][%d]\n", i, lane_act[i].cmd, lane_act[i].act);

    sprintf(wi.plz_id, "%.2s", PLZ_CODE2);
    nResult = lb_freadx(&wi, FNO_BASICTBL, &hi_comm_ctrl.bas_tbl, FFIRST);

    snprintf(szOrganizationID, 6+1, "%06d", lb_data2atoi(hi_comm_ctrl.bas_tbl.plz_id, 2));
lane_log("Packet[%s]\n", packet);
    snprintf(szWorkType, 4+1, "%04d", lane_act[i].work);

	sprintf(szData, "%s", packet);

    type = lane_act[i].act;
    job = lane_act[i].work;
//lane_log("szWorkType[%s] [%s]\n", szWorkType, lane_act[i].cmd_nm);
    snprintf(szMsgType       , 4+1, "%.4s", "0010");
lane_log("Web 처리할 자료[%s]\n", szData);
    //2019.06.27 InitDoubleLinkedList Main 선언
    InitThread_Before();
//lane_log("Queue 초기화\n");
    
	nResult = InitThread(ProcThreadT);

lane_log("Web 종료[%d]?[%s][%d]\n", type, rtnData, nResult);
    sprintf(rtn, "%s", rtnData);
    WriteLog(LogType_SystemOut,LogLevel_Info, "Process Stopped![%s]", g_szProcessname);

    return 0;
}

int InitMessageSenderConfigT(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);
//lane_log("Init 1\n");
    // 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;
    }
//lane_log("Init 2\n");
    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;
    }
//lane_log("Init 3\n");
    // 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;
    }
//lane_log("Init 4\n");
    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;
    }
//lane_log("Init 5\n");
    if (InitLogConfig() != LOG_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }
//lane_log("Init 6\n");
    if (InitSenderMQConfig() != MQ_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }
//lane_log("Init 7\n");
    if (InitSender2MQConfig() != MQ_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }
//lane_log("Init 8\n");
    if (InitThreadConfig() != THREAD_ERR_SUCCESS)
    {
        return MS_ERR_CONFIG;
    }
//lane_log("Init 9\n");
    return MS_ERR_SUCCESS;
}

void *ProcThreadT(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;
    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));
    int nMessageLength;
    //int nPacketLength;
    int nSerialNo;
    int nCompareResult;

    //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);
#if 0
                continue;
#else
	            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;
#endif
            }
            
            //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Connect SenderMQ is succeeded!!", pThreadInfo->nThreadIndex);
lane_log("Web Type [%d]\n", type);
		    snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendRealQueuename, szOrganizationID, pThreadInfo->nThreadIndex);
		    //snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendHqctQueuename, szOrganizationID, pThreadInfo->nThreadIndex);
        }

//lane_log("Reply Que 1 [%s]\n", szReplyQueuename);
//sleep(4);
        GetCurrTime(&tNow, &nMillisecond);
        nSerialNo = GetSerialNo();
//lane_log("Serial Que 2 [%d]\n", nSerialNo);
//sleep(4);
        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, szRespCodeT, procData);
        //free(szData);
        nMessageLength = strnlen_s(szMessageBuffer, sizeof(szMessageBuffer));
        sMessageInfo.nMessageLength = nMessageLength;
        sMessageInfo.paMessageData = NULL;
//lane_log("Aloc Before\n");
//sleep(4);
        SafeMemoryAlloc((void **)&sMessageInfo.paMessageData, sMessageInfo.nMessageLength + 5);
//lane_log("Aloc After\n");
//sleep(4);
        snprintf(sMessageInfo.paMessageData, sMessageInfo.nMessageLength + 5, "%04X", sMessageInfo.nMessageLength);
        SafeStrCat(sMessageInfo.paMessageData + 4, sMessageInfo.nMessageLength + 1, szMessageBuffer, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)));
//lane_log("A1\n");
//sleep(4);
        snprintf(szTopic, sizeof(szTopic), "%s.%s%s", szOrganizationID, szWorkType, szMsgType);
#ifdef _WIN32
        uuid_generate(&sUUIDTemp);
        uuid_unparse(&sUUIDTemp, szSendUUIDString);
#else
//lane_log("A2\n");
//sleep(4);
        uuid_generate(sUUIDTemp);
//lane_log("A3\n");
//sleep(4);
        uuid_unparse(sUUIDTemp, szSendUUIDString);
#endif
lane_log("Web TEST 2 Exchange[%s]\n", g_szMQMessageSendExchangername);
lane_log("Web TEST 3 Topic[%s]\n", szTopic);
lane_log("Web TEST 4 Reply[%s][%d]\n", szReplyQueuename, pThreadInfo->sPublishMQ.bMQChannelOpen);
//sleep(4);
        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);
lane_log("Put After\n");
//sleep(2);
        lane_log("Web SendData[%d] : [Thread:%02d]%s\n", nResult, pThreadInfo->nThreadIndex, sMessageInfo.paMessageData);
        if (nResult != MS_ERR_SUCCESS)
        {
			lane_log("Web [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);
			lane_log("Web [Thread:%02d]MQ Data Publish[Data:%s]!", pThreadInfo->nThreadIndex, 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! - UUID:%s, Message:%s, ProcessTime:%d.%03ds\n", pThreadInfo->nThreadIndex, szSendUUIDString,
                                                                      sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                        bTimeout = true;
                        break;
                    }

                    memset(rtnData, 0x00, sizeof(rtnData));
					memset(procData, 0x00, sizeof(procData));
                    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);
//lane_log("Get Msg[%s][%d]\n", szMessageBuffer, nResult);
                    snprintf(rtnData, strnlen_s(szMessageBuffer+45, sizeof(szMessageBuffer)-45)+1, "%s", szMessageBuffer+45);
lane_log("Web Rtn Data[%s]\n", rtnData);
                    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! - UUID:%s, Message:%s, ProcessTime:%d.%03ds\n",
                                 pThreadInfo->nThreadIndex, szSendUUIDString, sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                        bTimeout = true;
                        break;
                    }
printf("Web TEST11[%d][%d][%s]\n", type, nResult, rtnData);

                    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)
                {
                    // 전송 완료 처리 필요
                }else
                {
                    memset(rtnData, 0x00, sizeof(rtnData));
                    snprintf(rtnData, sizeof(rtnData), "%02x0002", cmd);
                }
            }
        }
        //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);
	pThreadInfo->nRunState = SOCK_THREAD_STATE_TERMINATED;
#if 1 //2019.07.12 메모리 증가로 Mutex Unlock 후 삭제 부분 추가
    pthread_mutex_destroy(&pThreadInfo->sContextMutex);
    pthread_mutex_destroy(&pThreadInfo->sPublishMQ.sMQMutex);
    pthread_mutex_destroy(&pThreadInfo->sConsumeMQ.sMQMutex);
#endif
    SafeFree((void **)&pThreadInfo);
    SafeFree((void **)&sMessageInfo.paMessageData);
    SafeFree((void **)&pnThreadExitCode);
    return 0;
}

int GetSerialNoT()
{
    int nSerialNo;
    pthread_mutex_lock(&g_sSerialMutexT);
    nSerialNo = g_nSerailNoT++;
    pthread_mutex_unlock(&g_sSerialMutexT);
    return nSerialNo;
}
