
#include "MessageProcessor.h"


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



int main(int argc, char *argv[])
{
    char szProcessname[PATH_MAX + 1];
    char *pFindPointer;
    //int nPathIndicatorPos;
    int nResult;

    DaemonStart();
    nResult = SafeStrCopy(szProcessname, sizeof(szProcessname), argv[0]);
    if (nResult != SEC_ERR_SUCCESS)
    {
        printf_s("strcpy_s Error![%d]\n", nResult);
    }
//    printf_s("Input argument : [%s]\n", argv[0]);
    pFindPointer = szProcessname;
    nResult = SafeStrrChr(szProcessname, strnlen_s(szProcessname, sizeof(szProcessname)), PATH_SEPERATOR, &pFindPointer);

//    printf_s("문자열길이 : %d\n", strnlen_s(szProcessname, sizeof(szProcessname)));
//    printf_s("szProcessname Address : %x\n", szProcessname);
//    printf_s("pFindPointer Address : %x\n", pFindPointer);
//    printf_s("szProcess Name : %s\n", szProcessname);
//    printf_s("g_szProcess Name : %s\n", g_szProcessname);

//    if (nResult != EOK)
//    {
//        printf_s("strrchr_s Error![%d]\n", nResult);
//    }

    if (pFindPointer == NULL)
    {
        nResult = SafeStrCopy(g_szProcessname, sizeof(g_szProcessname), szProcessname);
    }
    else
    {
        nResult = SafeStrCopy(g_szProcessname, sizeof(g_szProcessname), pFindPointer + 1);
    }
//    printf_s("szProcess Name : %s\n", szProcessname);
//    printf_s("g_szProcess Name : %s\n", g_szProcessname);

    GetExecPath(g_szProcessPath, sizeof(g_szProcessPath) - 1, argv[0]);


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

    nResult = InitMessageRelayConfig();

    if (nResult != MQ_ERR_SUCCESS)
    {
        printf_s("Program terminated!!!\n");
        return -1;
    }

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

    /* Socket 처리 루프 : 종료시 프로그램 종료됨 */
    nResult = InitThread(WorkThread);
    while(true)
    {
        sleep(5);
    }

    WriteLog(LogType_SystemOut, LogLevel_Info, "Process Stopped![%s]", g_szProcessname);

    return 0;

}


int InitMessageRelayConfig()
{
    char szConfigfilename[PATH_MAX + 1];
    char szConfigfilenameTemp[PATH_MAX + 1];
    char szValueTemp[MAX_STRING_SIZE + 1];
    int nResult;

//    snprintf(szConfigfilenameTemp, sizeof(szConfigfilenameTemp), "%s/%s/", g_szProcessPath, CONF_FILE_PATH);
//
//    realpath(szConfigfilenameTemp, szConfigfilename);
//
//    SafeStrCat(szConfigfilename, sizeof(szConfigfilename), "/", 1);
//    SafeStrCat(szConfigfilename, sizeof(szConfigfilename), g_szProcessname, strnlen_s(g_szProcessname, sizeof(g_szProcessname)));
//    SafeStrCat(szConfigfilename, sizeof(szConfigfilename), CONF_FILE_EXT, strnlen_s(CONF_FILE_EXT, sizeof(CONF_FILE_EXT)));
    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_s("Process Name : %s\n", g_szProcessname);
    printf_s("Process Path : %s\n", g_szProcessPath);
    printf_s("Config Filename : %s\n", szConfigfilename);

    ReadConfigFile(true, szConfigfilename);

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


    if (InitProcessorMQConfig() != MQ_ERR_SUCCESS)
    {
        return MP_ERR_CONFIG;
    }

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

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

    return MP_ERR_SUCCESS;
}

void *WorkThread(void *pData)
{
    struct THREAD_INFO *pThreadInfo;
    amqp_rpc_reply_t sRPCReply;
    char szMessageBuffer[MAX_STRING_SIZE + 1];
    int nResult;
    int *pnThreadExitCode;
    char szCorrelationID[MAX_STRING_SIZE + 1];
    char szReplyQueuename[MAX_STRING_SIZE + 1];
    char szExchangername[MAX_STRING_SIZE + 1];
    int nMessageLength;
    uint64_t nDeliveryTag;
    struct THREAD_INFO SendThreadInfo;

    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);
    //SendThreadInfo.nRunState = THREAD_STATE_INIT;
    //pthread_mutex_lock(&pThreadInfo->sContextMutex);
    SendThreadInfo.nThreadIndex = pThreadInfo->nThreadIndex;
    SendThreadInfo.bThreadLoop = true;

    pnThreadExitCode = NULL;
    nResult = SafeMemoryAlloc((void **)&pnThreadExitCode, sizeof(int));
    //pnThreadExitCode = SafeAlloc(pnThreadExitCode, sizeof(int));
    *pnThreadExitCode = 0;
    while (pThreadInfo->bThreadLoop)
    {
        if (pThreadInfo->sPublishMQ.bMQConnection == false || pThreadInfo->sPublishMQ.bMQChannelOpen == false
            || pThreadInfo->sConsumeMQ.bMQConnection == false || pThreadInfo->sConsumeMQ.bMQChannelOpen == false)
        {
            if (ConnectProcessorMQ(pThreadInfo, true) != MQ_ERR_SUCCESS)
            {
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]ConnectProcessMQ is failed!", pThreadInfo->nThreadIndex);
                continue;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]ConnectProcessMQ is succeeded!!", pThreadInfo->nThreadIndex);
        }
        if (SendThreadInfo.sPublishMQ.bMQConnection == false || SendThreadInfo.sPublishMQ.bMQChannelOpen == false
            || SendThreadInfo.sConsumeMQ.bMQConnection == false || SendThreadInfo.sConsumeMQ.bMQChannelOpen == false)
        {
            if (ConnectSenderMQ(&SendThreadInfo, true) != MQ_ERR_SUCCESS)
            {
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]ConnectSenderMQ is failed!", SendThreadInfo.nThreadIndex);
                continue;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]ConnectSenderMQ is succeeded!!", SendThreadInfo.nThreadIndex);
        }
        //amqp_maybe_release_buffers(pThreadInfo->sMQConnectionState);
        nResult = GetMessageFromQueue(pThreadInfo->sConsumeMQ.sMQConnectionState, szExchangername, sizeof(szExchangername), szMessageBuffer, sizeof(szMessageBuffer), szCorrelationID, sizeof(szCorrelationID), szReplyQueuename, sizeof(szReplyQueuename), &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen, &nDeliveryTag, NULL);
        nMessageLength = strnlen_s(szMessageBuffer, sizeof(szMessageBuffer));
        if (nResult == MQ_ERR_SUCCESS)
        {
            //printf_s("Message Received : %s\n", szMessageBuffer);
            //WriteRawCommLog(LogType_CommOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'C', "%s", szMessageBuffer);
            WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'C', g_szMQMessageRecvQueuename, "", szCorrelationID, nMessageLength, "%s", szMessageBuffer);
            WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Consume[Data:%s]!", pThreadInfo->nThreadIndex, szMessageBuffer);
            //nResult = PutAckMessageToQueue(pThreadInfo->sMQConnectionState, szCorrelationID, szReplyQueuename, szMessageBuffer, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)), &pThreadInfo->bMQConnection, &pThreadInfo->bChannelOpen);
            
            // 메시지 DB 처리 등 처리 - Relay서버는 bypass

            GetCurrTime(&tNow, &nMillisecond);
            nSerialNo = GetSerialNo();
            // MakeMSGFrame함수 사용하게 변경 필요
            snprintf(szData, sizeof(szData), "Hello%08d!", nSerialNo);
            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, szData);

            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)));
            //nPacketLength = 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(SendThreadInfo.sPublishMQ.sMQConnectionState, &SendThreadInfo.sPublishMQ.bMQConnection, &SendThreadInfo.sPublishMQ.bMQChannelOpen, SendThreadInfo.nThreadIndex, g_szMQMessageSendExchangername, szReplyQueuename, szTopic, &sMessageInfo, sMessageInfo.paMessageData, sMessageInfo.nMessageLength, szSendUUIDString, sizeof(szSendUUIDString), false, NULL);
            //printf_s("SendData : [Thread:%02d]%s\n", SendThreadInfo.nThreadIndex, sMessageInfo.paMessageData);

            if (nResult != MS_ERR_SUCCESS)
            {
                // 메시지 전송 오류인 경우 처리
                // 오류인 경우 오류 로그에 추가하는 형태의 샘플이지만 DB 처리라면 재전송 여부를 결정해서 재전송 몇번 이내면 재전송 하고 그렇지 않으면 실패 처리하는 로직이 필요하다.
                WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Publish Error[Data:%s]!", SendThreadInfo.nThreadIndex, sMessageInfo.paMessageData);
            }
            else
            {
                WriteQueueLog(LogType_MQOut, LogLevel_Info, SendThreadInfo.nThreadIndex, 'P', g_szMQMessageSendExchangername, szTopic, szSendUUIDString, sMessageInfo.nMessageLength, "%s", sMessageInfo.paMessageData);
                WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Publish[Data:%s]!", SendThreadInfo.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", SendThreadInfo.nThreadIndex, szSendUUIDString, sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                            bTimeout = true;
                            break;
                        }
                        nResult = GetMessageFromQueue(SendThreadInfo.sConsumeMQ.sMQConnectionState, szExchangername, sizeof(szExchangername), szMessageBuffer, sizeof(szMessageBuffer), szRecvUUIDString, sizeof(szRecvUUIDString), szReplyQueuename, sizeof(szReplyQueuename), &SendThreadInfo.sConsumeMQ.bMQConnection, &SendThreadInfo.sConsumeMQ.bMQChannelOpen, &nDeliveryTag, &sResponseTimeout);
                        // Response Timeout Process
                        // 메시지 전송 방식에 맞게 프로젝트별 수정 필요
                        // 이미 전송했던 전문을 재전송 할 것인지?, 업무 프로세스상 오류로 처리 하고 다시 처리 할 것인지?
                        // 타임아웃이 발생하였을 경우 DB에 처리 할 것인지 등등의 처리 방법에 따라 처리
                        // 지속적으로 전송하는 프로세스와 특정 시간에 실행하여 전송하는 형태가 필요할 것으로 판단됨
                        // 배치성 메시지 전송이라면 복잡한 쓰레드 없이 단순히 반복 루프만으로 처리 가능
            //            clock_gettime(CLOCK_REALTIME, &sRecvTimeSpec);
            //            sRecvNow = sRecvTimeSpec.tv_sec;
            //            nRecvMillisecond = sRecvTimeSpec.tv_nsec / 1.0e6; // nanoseconds to milliseconds
            //
            //            nDiffSec = sRecvNow - sNow;
            //            nDiffMilliSec = nRecvMillisecond - nMillisecond;
                        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_s("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", SendThreadInfo.nThreadIndex, szSendUUIDString, sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                            bTimeout = true;

                            snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s%s%s%s%s%04d%02d%02d%02d%02d%02d%s",
                                     g_szMQOrganizationCode, cOccurType, szWorkType, "0020", rSerial,
                                     tNow.tm_year + 1900, tNow.tm_mon + 1, tNow.tm_mday, tNow.tm_hour, tNow.tm_min, tNow.tm_sec,
                                     "0002");

                            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", SendThreadInfo.nThreadIndex, szSendUUIDString, szRecvUUIDString, sMessageInfo.paMessageData, nDiffSec, nDiffMilliSec);
                            amqp_basic_ack(SendThreadInfo.sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
                            continue;
                        }
                        bMessageMatch = true;
                        nMessageLength = strnlen_s(szMessageBuffer, sizeof(szMessageBuffer));
                        amqp_basic_ack(SendThreadInfo.sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
                        WriteQueueLog(LogType_MQOut, LogLevel_Info, SendThreadInfo.nThreadIndex, 'C', szReplyQueuename, "", szRecvUUIDString, nMessageLength, "%s", szMessageBuffer);
                        WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Consume[Data:%s] - ProcessTime:%d.%03ds!", SendThreadInfo.nThreadIndex, szMessageBuffer, nDiffSec, nDiffMilliSec);
                    } while (bMessageMatch == false && bTimeout == false);
                    if (!bTimeout)
                    {
                        // 전송 완료 처리 필요
                    }
                }
            }

            if (g_bBypassMode)
            {
		        amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
                nResult = PutReplyMessageToQueue(pThreadInfo->sPublishMQ.sMQConnectionState, szCorrelationID, szReplyQueuename, szMessageBuffer, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)), &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
                WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'P', "", szReplyQueuename, szCorrelationID, nMessageLength, "%s", szMessageBuffer);
                WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]Response Publish[Queue:%s,UUID:%s,Data:%s]!", pThreadInfo->nThreadIndex, szReplyQueuename, szCorrelationID, szMessageBuffer);
            }
            else
          	{
                if (/* 오류 판단*/)
          		{
                    amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
                }
                else
                {
                    amqp_basic_nack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0, 1);
		      	}
          	}
        }
        else
        {
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]MQ Data Consume Error!", pThreadInfo->nThreadIndex);
        }
    }
    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);
    return pnThreadExitCode;
}

