/*
#include <amqp.h>
#include <amqp_tcp_socket.h>
#include <amqp_framing.h>
#include <limits.h>
#include <stdbool.h>
#include <pthread.h>

#include "Common.h"
#include "DataStructure.h"
#include "Config.h"
#include "Log.h"
#include "StringUtil.h"
#include "Socket.h"
*/
#include "MessageQueue.h"

char g_szMQServerIP[MAX_STRING_SIZE + 1];
int g_nMQServerPort;
char g_szMQVirtualHostname[MAX_STRING_SIZE + 1];
char g_szMQServiceID[MAX_STRING_SIZE + 1];
char g_szMQServicePWD[MAX_STRING_SIZE + 1];
char g_szMQMessageSendQueuename[MAX_STRING_SIZE + 1];
char g_szMQMessageSendRespQueuename[MAX_STRING_SIZE + 1];
char g_szMQMessageRecvQueuename[MAX_STRING_SIZE + 1];
char g_szMQMessageRecvRespQueuename[MAX_STRING_SIZE + 1];
#ifdef _ETCS_ //응답큐 추가(by JBY)
char g_szMQMessageSendRealQueuename[MAX_STRING_SIZE + 1];
char g_szMQMessageSendHqctQueuename[MAX_STRING_SIZE + 1];
#endif
char g_szMQMessageSendExchangername[MAX_STRING_SIZE + 1];
#ifdef _ETCS_ //송신 Exchange 추가(by JBY)
char g_szMQMessageSend2Exchangername[MAX_STRING_SIZE + 1];
#endif
char g_szMQMessageRecvExchangername[MAX_STRING_SIZE + 1];
char g_szMQFileSendQueuename[MAX_STRING_SIZE + 1];
char g_szMQFileSendRespQueuename[MAX_STRING_SIZE + 1];
char g_szMQFileRecvQueuename[MAX_STRING_SIZE + 1];
char g_szMQFileSendExchangername[MAX_STRING_SIZE + 1];
char g_szMQFileRecvExchangername[MAX_STRING_SIZE + 1];
char g_szMQTargetOrganizationCode[MAX_STRING_SIZE + 1];
char g_szMQOrganizationCode[MAX_STRING_SIZE + 1];

int g_nPrefetchSize;
int g_nPrefetchCount;
int g_bBypassMode = true;

pthread_mutex_t g_sMQMutex = PTHREAD_MUTEX_INITIALIZER;


int g_nProcessType;

struct VALUE_STRING_PAIR_INFO g_sProcTypeInfo[] =
{
    { MQ_PROC_TYPE_GW,              "Gateway"           },
    { MQ_PROC_TYPE_PROCESSOR,       "Processor"         },
    { MQ_PROC_TYPE_SENDER,          "Sender"            },
    { MQ_PROC_TYPE_RELAY,           "Relay"             },
};


/*
 * topic의 경우 기관코드.전문코드.socket 번호
 * 송신큐의 경우 라우팅 문제로 큐에 기관코드가 할당되어 있으며 수신 응답 큐(Send 큐)의 경우 소켓번호까지 할당되어 있다.
 * 송신큐와 수신응답 큐는 분리되어 관리된다.
 *
 * rabbitMQ도 fsync를 이용하면 매우 느려진다는 글이 보인다. 테스트가 필요하다.
 * fsync를 이용하면서 수신시 DB 처리전에 응답을 주는 경우에는 서버 종료시 전문 누락이 가능하다.
 *
 * consumer 등록 하고 message를 consume 한다.
 *
 * 응답이 필요한 경우 correlation id를 uuid_generate를 이용하여 할당 하고 다시 이를 돌려 받아 처리한다.
 * correlation id가 필요한 이유는 이미 타임아웃 처리된 전문이 나중에 다시 응답 받을 수도 있어 통신 순서문제로 인한 문제가 발생하지 않게 하기 위함이다.
 *
 * amqp_frames_enqueued 큐에 데이터 여부 확인 함수
 *
 * amqp_queue_declare
 *
 * amqp_queue_bind
 *
 * amqp_exchange_declare
 *
 */

int InitMQConfig()
{
    char szValueTemp[MAX_STRING_SIZE + 1];
/*
#define MQ_CONF_SERVER_IP_KEY                   "MQ_SERVER_IP"
#define MQ_CONF_SERVER_PORT_KEY                 "MQ_SERVER_PORT"
#define MQ_CONF_VIRTUALHOST_KEY                 "MQ_VIRTUALHOST"
#define MQ_CONF_SERVICE_USER_ID_KEY             "MQ_SERVICE_USER_ID"
#define MQ_CONF_SERVICE_PWD_KEY                 "MQ_SERVICE_PWD"
#define MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY     "MQ_MESSAGE_SEND_QUEUE_NAME"
#define MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY "MQ_MESSAGE_RECV_EXCHANGER_NAME"
#define MQ_CONF_FILE_SEND_QUEUE_NAME_KEY        "MQ_FILE_SEND_QUEUE_NAME"
#define MQ_CONF_FILE_RECV_EXCHANER_NAME_KEY     "MQ_FILE_RECV_EXCHANER_NAME"
*/

    // Read MQ_CONF_SERVER_IP_KEY
    if (GetConfigValue(MQ_CONF_SERVER_IP_KEY, g_szMQServerIP, sizeof(g_szMQServerIP)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_IP_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVER_PORT_KEY
    if (GetConfigValue(MQ_CONF_SERVER_PORT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_PORT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nMQServerPort) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_SERVER_PORT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_VIRTUALHOST_KEY
    if (GetConfigValue(MQ_CONF_VIRTUALHOST_KEY, g_szMQVirtualHostname, sizeof(g_szMQVirtualHostname)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_VIRTUALHOST_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_USER_ID_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_USER_ID_KEY, g_szMQServiceID, sizeof(g_szMQServiceID)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_USER_ID_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_PWD_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_PWD_KEY, g_szMQServicePWD, sizeof(g_szMQServicePWD)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_PWD_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY, g_szMQMessageSendExchangername, sizeof(g_szMQMessageSendExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY, g_szMQMessageRecvExchangername, sizeof(g_szMQMessageRecvExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY, g_szMQMessageSendQueuename, sizeof(g_szMQMessageSendQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY, g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }
    // Read MQ_CONF_MESSAGE_RECV_RESP_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_RESP_QUEUE_NAME_KEY, g_szMQMessageRecvRespQueuename, sizeof(g_szMQMessageRecvRespQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_RESP_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_FILE_SEND_EXCHANER_NAME_KEY
    if (GetConfigValue(MQ_CONF_FILE_SEND_EXCHANER_NAME_KEY, g_szMQFileSendExchangername, sizeof(g_szMQFileSendExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_FILE_SEND_EXCHANER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_FILE_RECV_EXCHANER_NAME_KEY
    if (GetConfigValue(MQ_CONF_FILE_RECV_EXCHANER_NAME_KEY, g_szMQFileRecvExchangername, sizeof(g_szMQFileRecvExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_FILE_RECV_EXCHANER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_FILE_SEND_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_FILE_SEND_QUEUE_NAME_KEY, g_szMQFileSendQueuename, sizeof(g_szMQFileSendQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_FILE_SEND_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_FILE_RECV_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_FILE_RECV_QUEUE_NAME_KEY, g_szMQFileRecvQueuename, sizeof(g_szMQFileRecvQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_FILE_RECV_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }


    // Read MQ_CONF_TARGET_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_TARGET_ORGANIZATION_CODE, g_szMQTargetOrganizationCode, sizeof(g_szMQTargetOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_TARGET_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_ORGANIZATION_CODE, g_szMQOrganizationCode, sizeof(g_szMQOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }
#if 0 
//현재 라이브러리 버젼은 TARGET으로 변경됨.
//하지만 차로로부터 수신 받아서 DB처리 후 외부 송신 위해서 ORGINAL 처리로 남김
    // Read MQ_CONF_TARGET_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_TARGET_ORGANIZATION_CODE, g_szMQTargetOrganizationCode, sizeof(g_szMQTargetOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }
#endif
    // Read PROCESS_TYPE
    if (GetConfigValue(MQ_CONF_PROCESS_TYPE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration not found[Available Value:Gateway or Processor]!");
        return MQ_ERR_CONFIG;
    }
    if (MappingStringToInt(szValueTemp, sizeof(szValueTemp), g_sProcTypeInfo, sizeof(g_sProcTypeInfo)/sizeof(struct VALUE_STRING_PAIR_INFO), &g_nProcessType) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration error[ConfigFile:%s,Available Value:Gateway or Processor]!", szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_SIZE_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_SIZE_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchSize) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_COUNT_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_COUNT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchCount) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read BYPASS_MODE
    if (GetConfigValue(SOCK_CONF_BYPASS_MODE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        g_bBypassMode = true;
    }
    else
    {
        if (ParsingBoolean(szValueTemp, sizeof(szValueTemp), &g_bBypassMode) != CONF_ERR_SUCCESS)
        {
            g_bBypassMode = true;
        }
    }

    return MQ_ERR_SUCCESS;
}

int InitProcessorMQConfig()
{
    char szValueTemp[MAX_STRING_SIZE + 1];
/*
MQ_SERVER_IP=127.0.0.1
MQ_SERVER_PORT=5672
MQ_VIRTUALHOST=MessageService
#MQ_SERVICE_USER_ID=siadmin
MQ_SERVICE_USER_ID=daservice
MQ_SERVICE_PWD=neighbor21
MQ_MESSAGE_SEND_EXCHANGER_NAME=MessageSencExc
MQ_MESSAGE_RECV_QUEUE_NAME=MessageRecvQ
*/

    // Read MQ_CONF_SERVER_IP_KEY
    if (GetConfigValue(MQ_CONF_SERVER_IP_KEY, g_szMQServerIP, sizeof(g_szMQServerIP)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_IP_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVER_PORT_KEY
    if (GetConfigValue(MQ_CONF_SERVER_PORT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_PORT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nMQServerPort) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_SERVER_PORT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_VIRTUALHOST_KEY
    if (GetConfigValue(MQ_CONF_VIRTUALHOST_KEY, g_szMQVirtualHostname, sizeof(g_szMQVirtualHostname)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_VIRTUALHOST_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_USER_ID_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_USER_ID_KEY, g_szMQServiceID, sizeof(g_szMQServiceID)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_USER_ID_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_PWD_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_PWD_KEY, g_szMQServicePWD, sizeof(g_szMQServicePWD)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_PWD_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY, g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_ORGANIZATION_CODE, g_szMQOrganizationCode, sizeof(g_szMQOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }

    // Read PROCESS_TYPE
    if (GetConfigValue(MQ_CONF_PROCESS_TYPE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration not found[Available Value:Gateway or Application]!");
        return MQ_ERR_CONFIG;
    }
    if (MappingStringToInt(szValueTemp, sizeof(szValueTemp), g_sProcTypeInfo, sizeof(g_sProcTypeInfo)/sizeof(struct VALUE_STRING_PAIR_INFO), &g_nProcessType) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration error[ConfigFile:%s,Available Value:Server or Client]!", szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_SIZE_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_SIZE_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchSize) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_COUNT_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_COUNT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchCount) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read BYPASS_MODE
    if (GetConfigValue(SOCK_CONF_BYPASS_MODE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        g_bBypassMode = true;
    }
    else
    {
        if (ParsingBoolean(szValueTemp, sizeof(szValueTemp), &g_bBypassMode) != CONF_ERR_SUCCESS)
        {
            g_bBypassMode = true;
        }
    }

    return MQ_ERR_SUCCESS;
}

int InitProcessor2MQConfig()
{
    // Read MQ_CONF_MESSAGE_SEND_RESP_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_RESP_QUEUE_NAME_KEY, g_szMQMessageSendRespQueuename, sizeof(g_szMQMessageSendRespQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_RESP_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY, g_szMQMessageSendExchangername, sizeof(g_szMQMessageSendExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }
    return MQ_ERR_SUCCESS;
}

int InitSenderMQConfig()
{
    char szValueTemp[MAX_STRING_SIZE + 1];
/*
MQ_SERVER_IP=127.0.0.1
MQ_SERVER_PORT=5672
MQ_VIRTUALHOST=MessageService
#MQ_SERVICE_USER_ID=siadmin
MQ_SERVICE_USER_ID=daservice
MQ_SERVICE_PWD=neighbor21
MQ_MESSAGE_SEND_EXCHANGER_NAME=MessageSencExc
MQ_MESSAGE_RECV_QUEUE_NAME=MessageRecvQ
*/

    // Read MQ_CONF_SERVER_IP_KEY
    if (GetConfigValue(MQ_CONF_SERVER_IP_KEY, g_szMQServerIP, sizeof(g_szMQServerIP)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_IP_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVER_PORT_KEY
    if (GetConfigValue(MQ_CONF_SERVER_PORT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_PORT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nMQServerPort) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_SERVER_PORT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_VIRTUALHOST_KEY
    if (GetConfigValue(MQ_CONF_VIRTUALHOST_KEY, g_szMQVirtualHostname, sizeof(g_szMQVirtualHostname)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_VIRTUALHOST_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_USER_ID_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_USER_ID_KEY, g_szMQServiceID, sizeof(g_szMQServiceID)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_USER_ID_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_PWD_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_PWD_KEY, g_szMQServicePWD, sizeof(g_szMQServicePWD)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_PWD_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY -> Changed to SEND_RESP_QUEUE
//    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY, g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename)) != MQ_ERR_SUCCESS)
//    {
//        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY);
//        return MQ_ERR_CONFIG;
//    }


    // Read MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY
//    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY, g_szMQMessageSendQueuename, sizeof(g_szMQMessageSendQueuename)) != MQ_ERR_SUCCESS)
//    {
//        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_QUEUE_NAME_KEY);
//        return MQ_ERR_CONFIG;
//    }

    // Read MQ_CONF_MESSAGE_SEND_RESP_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_RESP_QUEUE_NAME_KEY, g_szMQMessageSendRespQueuename, sizeof(g_szMQMessageSendRespQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_RESP_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY, g_szMQMessageSendExchangername, sizeof(g_szMQMessageSendExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_ORGANIZATION_CODE, g_szMQOrganizationCode, sizeof(g_szMQOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }

    // Read PROCESS_TYPE
    if (GetConfigValue(MQ_CONF_PROCESS_TYPE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration not found[Available Value:Gateway or Application]!");
        return MQ_ERR_CONFIG;
    }
    if (MappingStringToInt(szValueTemp, sizeof(szValueTemp), g_sProcTypeInfo, sizeof(g_sProcTypeInfo)/sizeof(struct VALUE_STRING_PAIR_INFO), &g_nProcessType) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration error[ConfigFile:%s,Available Value:Server or Client]!", szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_SIZE_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_SIZE_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchSize) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_COUNT_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_COUNT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchCount) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read BYPASS_MODE
    if (GetConfigValue(SOCK_CONF_BYPASS_MODE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        g_bBypassMode = true;
    }
    else
    {
        if (ParsingBoolean(szValueTemp, sizeof(szValueTemp), &g_bBypassMode) != CONF_ERR_SUCCESS)
        {
            g_bBypassMode = true;
        }
    }

    return MQ_ERR_SUCCESS;
}
#ifdef _ETCS_
int InitSender2MQConfig()
{
    // Read MQ_CONF_MESSAGE_SEND_REAL_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_REAL_QUEUE_NAME_KEY, g_szMQMessageSendRealQueuename, sizeof(g_szMQMessageSendRealQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_REAL_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_SEND_HQCT_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND_HQCT_QUEUE_NAME_KEY, g_szMQMessageSendHqctQueuename, sizeof(g_szMQMessageSendHqctQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_HQCT_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_SEND2_EXCHANGER_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_SEND2_EXCHANGER_NAME_KEY, g_szMQMessageSend2Exchangername, sizeof(g_szMQMessageSend2Exchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_SEND_EXCHANGER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }
    
    // Read MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY, g_szMQMessageRecvExchangername, sizeof(g_szMQMessageRecvExchangername)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_EXCHANGER_NAME_KEY);
        return MQ_ERR_CONFIG;
    }
    
    // Read MQ_CONF_TARGET_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_TARGET_ORGANIZATION_CODE, g_szMQTargetOrganizationCode, sizeof(g_szMQTargetOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_TARGET_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }
    return MQ_ERR_SUCCESS;
}
#endif
void AMQPStatusProcess(int nStatus, const char *szMessage)
{
    char szMessageBuffer[MAX_STRING_SIZE + 1];
    /*
     * TODO : 각 상황에 따라 추가 처리가 필요한지 메시지별 확인 필요(Document에 아무런 설명이 없음)
     *        socket 오류의 경우 무시하면 되는지?, 연결을 끊어야 되는지 확인이 필요함
     *        각 함수에서 발생 가능한 오류를 명시해야 하지만 확인이 안됨
     */
    switch (nStatus)
    {

    case AMQP_STATUS_NO_MEMORY:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Memory allocation failed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_BAD_AMQP_DATA:
        //This is a protocol error.
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Incorrect or corrupt data was received from the broker.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_UNKNOWN_CLASS:
        //This is a protocol error.
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:An unknown AMQP class was received.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_UNKNOWN_METHOD:
        //This is a protocol error.
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:An unknown AMQP method was received.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_HOSTNAME_RESOLUTION_FAILED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Unable to resolve the hostname.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_INCOMPATIBLE_AMQP_VERSION:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:The broker advertised an incompaible AMQP version.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_CONNECTION_CLOSED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:The connection to the broker has been closed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_BAD_URL:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:malformed AMQP URL]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SOCKET_ERROR:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:A socket error occurred.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_INVALID_PARAMETER:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:An invalid parameter was passed into the function.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_TABLE_TOO_BIG:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:The amqp_table_t object cannot be serialized because the output buffer is too small.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_WRONG_METHOD:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:The wrong method was received.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_TIMEOUT:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Operation timed out.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_TIMER_FAILURE:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:The underlying system timer facility failed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_HEARTBEAT_TIMEOUT:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Timed out waiting for heartbeat.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_UNEXPECTED_STATE:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Unexpected protocol state.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SOCKET_CLOSED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Underlying socket is closed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SOCKET_INUSE:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Underlying socket is already open.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_BROKER_UNSUPPORTED_SASL_METHOD:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Broker does not support the requested SASL mechanism.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_UNSUPPORTED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Parameter is unsupported in this version.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case _AMQP_STATUS_NEXT_VALUE:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Internal value - _AMQP_STATUS_NEXT_VALUE.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_TCP_ERROR:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:A generic TCP error occurred.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_TCP_SOCKETLIB_INIT_ERROR:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:An error occurred trying to initialize the socket library.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case _AMQP_STATUS_TCP_NEXT_VALUE:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Internal value - _AMQP_STATUS_TCP_NEXT_VALUE.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SSL_ERROR:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:A generic SSL error occurred.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SSL_HOSTNAME_VERIFY_FAILED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:SSL validation of hostname against peer certificate failed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SSL_PEER_VERIFY_FAILED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:SSL validation of peer certificate failed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case AMQP_STATUS_SSL_CONNECTION_FAILED:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:SSL handshake failed.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    case _AMQP_STATUS_SSL_NEXT_VALUE:
        snprintf(szMessageBuffer, sizeof(szMessageBuffer), "%s[Status:Internal value - _AMQP_STATUS_SSL_NEXT_VALUE.]",szMessage);
        WriteLog(LogType_StandardOut, LogLevel_Info, szMessageBuffer);
        break;
    }
}

int AMQPReplyProcess(amqp_rpc_reply_t *sAMQPReply, const char *szString, int *pbMQConnection, int *pbMQChannelOpen)
{
    amqp_connection_close_t *pMQConnectionClose;
    amqp_channel_close_t *pMQChannelClose;
    switch (sAMQPReply->reply_type)
    {
    case AMQP_RESPONSE_NORMAL:
        return MQ_ERR_SUCCESS;

    case AMQP_RESPONSE_NONE:
        WriteLog(LogType_ErrorOut, LogLevel_Error, "%s: missing RPC reply type!\n", szString);
        return MQ_ERR_NO_RESPONSE;

    case AMQP_RESPONSE_LIBRARY_EXCEPTION:
        WriteLog(LogType_ErrorOut, LogLevel_Error, "%s: %s\n", szString, amqp_error_string2(sAMQPReply->library_error));
        return MQ_ERR_LIB_EXCEPTION;

    case AMQP_RESPONSE_SERVER_EXCEPTION:
        switch (sAMQPReply->reply.id)
        {
        case AMQP_CONNECTION_CLOSE_METHOD:
        {
            *pbMQConnection = false;
            pMQConnectionClose = (amqp_connection_close_t *)sAMQPReply->reply.decoded;
            WriteLog(LogType_ErrorOut, LogLevel_Error, "%s: server connection error %uh, message: %.*s\n",
                    szString, pMQConnectionClose->reply_code, (int)pMQConnectionClose->reply_text.len,
                    (char *)pMQConnectionClose->reply_text.bytes);
            break;
        }
        case AMQP_CHANNEL_CLOSE_METHOD:
        {
            *pbMQChannelOpen = false;
            pMQChannelClose = (amqp_channel_close_t *)sAMQPReply->reply.decoded;
            WriteLog(LogType_ErrorOut, LogLevel_Error, "%s: server channel error %uh, message: %.*s\n",
                    szString, pMQChannelClose->reply_code, (int)pMQChannelClose->reply_text.len,
                (char *)pMQChannelClose->reply_text.bytes);
            break;
        }
        default:
            WriteLog(LogType_ErrorOut, LogLevel_Error, "%s: unknown server error, method id 0x%08X\n",
                    szString, sAMQPReply->reply.id);
            break;
        }
        return MQ_ERR_SERVER_EXCEPTION;
    }
    return MQ_ERR_UNKNOWN;
}

int InitMQ(struct SOCKET_THREAD_INFO *pThreadInfo)
{
    pthread_mutex_init(&pThreadInfo->sPublishMQ.sMQMutex, NULL);
    pthread_mutex_lock(&pThreadInfo->sPublishMQ.sMQMutex);
    pThreadInfo->sPublishMQ.sMQConnectionState = amqp_new_connection();

    if (pThreadInfo->sPublishMQ.sMQConnectionState == NULL)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't Initialize Publish MessageQueue! Check Configuration!!!");
        pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
        return MQ_ERR_INIT;
    }
    pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);

    pthread_mutex_init(&pThreadInfo->sConsumeMQ.sMQMutex, NULL);
    pthread_mutex_lock(&pThreadInfo->sConsumeMQ.sMQMutex);
    pThreadInfo->sConsumeMQ.sMQConnectionState = amqp_new_connection();

    if (pThreadInfo->sConsumeMQ.sMQConnectionState == NULL)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't Initialize Consume MessageQueue! Check Configuration!!!");
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_INIT;
    }
    pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
return MQ_ERR_SUCCESS;
}

int ConnectSocketMQ(struct SOCKET_THREAD_INFO *pThreadInfo)
{
    int nResult;
    amqp_rpc_reply_t sAMQPReply;
    amqp_queue_declare_ok_t *pQueueDeclareResult;
    amqp_queue_purge_ok_t *pQueuePurgeResult;
    int nChannelNo;
    int nTypeFlags;
    char szStringBuffer[MAX_STRING_SIZE + 1];
    char szTopic[MAX_STRING_SIZE + 1];



    /*
     * 아래 두 변수를 이용하여 사용하는 큐를 확정하고 이를 초기화 해야 한다.
     * Message 수신을 사용하면 수신 큐, 수신 응답 큐 바인딩
     * Message 송신을 사용하면 송신 큐, 송신
    int g_nChannelCount = 0;
    int g_nChannelTypeFlags[SOCK_CONF_MAX_CHANNEL_LIMIT];

    SOCK_CHANNEL_MESSAGE_SEND   = 0x01,
    SOCK_CHANNEL_MESSAGE_RECV   = 0x02,
    SOCK_CHANNEL_FILE_SEND      = 0x04,
    SOCK_CHANNEL_FILE_RECV      = 0x08,
    */

    nTypeFlags = 0;

    for (nChannelNo = 0; nChannelNo < g_nChannelCount; nChannelNo++)
    {
        nTypeFlags |= g_nChannelTypeFlags[nChannelNo];
    }


    // Connect Publish MQ

    pthread_mutex_lock(&pThreadInfo->sPublishMQ.sMQMutex);

    if (pThreadInfo->sPublishMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sPublishMQ.sMQConnectionState = amqp_new_connection();
        //}

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sPublishMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sPublishMQ.sMQConnectionState);
            if (pThreadInfo->sPublishMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Message Queue socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
        //}

        nResult = amqp_socket_open(pThreadInfo->sPublishMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */


        sAMQPReply = amqp_login(pThreadInfo->sPublishMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Message Queue!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            break;
        case MQ_ERR_LIB_EXCEPTION:
            pThreadInfo->sPublishMQ.bMQConnection = false;
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            pThreadInfo->sPublishMQ.bMQConnection = false;
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sPublishMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sPublishMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
        AMQPReplyProcess(&sAMQPReply, "amqp_channel_open error[Publish채널]!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        return MQ_ERR_OPEN_CHANNEL;
    }
    pThreadInfo->sPublishMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sPublishMQ.sMQConnectionState,1);

    amqp_basic_qos(pThreadInfo->sPublishMQ.sMQConnectionState, 1, g_nPrefetchSize, g_nPrefetchCount, 0);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
        AMQPReplyProcess(&sAMQPReply, "amqp_basic_qos error[Publish채널]!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        return MQ_ERR_OPEN_CHANNEL;
    }


    if ((nTypeFlags & SOCK_CHANNEL_MESSAGE_RECV) != 0)
    {
        // Publish
        // 메시지 수신 익스체인저
        amqp_exchange_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQMessageRecvExchangername),
                              amqp_cstring_bytes("topic"), 0, 1, 0, 0,
                              amqp_empty_table);
        // declare 오류 체크 방법 확인 필요(Example에는 설명 없음, 문서에도 설명 없음)
        // 메시지 수신 큐
        snprintf(szStringBuffer, sizeof(szStringBuffer), "%s%s", g_szMQMessageRecvQueuename, g_szMQOrganizationCode);
        amqp_queue_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);
        // 메시지 수신 응답 큐 -> Consumer로 이동
//        snprintf(szStringBuffer, sizeof(szStringBuffer), "%s.%02d", g_szMQMessageSendQueuename, pThreadInfo->nSocketIndex);
//        pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);
//        if (pQueueDeclareResult->message_count > 0)
//        {
//            pQueuePurgeResult = amqp_queue_purge(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer));
//            WriteLog(LogType_StandardOut, LogLevel_Info, "MQ Purged[Queuename:%s, MessageCount:%d]!", szStringBuffer, pQueuePurgeResult->message_count);
//        }
        // 메시지 수신 바인딩
        snprintf(szTopic, sizeof(szTopic), "%s.*", g_szMQOrganizationCode);
        amqp_queue_bind(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer),
                        amqp_cstring_bytes(g_szMQMessageRecvExchangername), amqp_cstring_bytes(szTopic),
                        amqp_empty_table);
        sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error[메시지 수신 바인딩]!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        }

        // 메시지 수신 응답 바인딩 (자바 클라이언트에서 direct의 경우 exchanger를 ""로 하여 topic을 큐 이름으로 주고 처리)
//        amqp_queue_bind(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer),
//                        amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(szStringBuffer),
//                        amqp_empty_table);
//        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
//        {
//            AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error[메시지 수신 응답 바인딩]!", pThreadInfo->sPublishMQ.sMQConnectionState, &pThreadInfo->sPublishMQ.bMQChannelOpen);
//        }
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Unbinding");

        // Consume
        // 메시지 수신 응답 큐 Consume
//        amqp_basic_consume(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), amqp_empty_bytes, 0, 0, 0,
//                           amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
    }

    if ((nTypeFlags & SOCK_CHANNEL_FILE_RECV) != 0)
    {
        amqp_exchange_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileRecvExchangername),
                              amqp_cstring_bytes("topic"), 0, 1, 0, 0,
                              amqp_empty_table);

        snprintf(szStringBuffer, sizeof(szStringBuffer), "%s%s", g_szMQFileRecvQueuename, g_szMQOrganizationCode);
        amqp_queue_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);
        //snprintf(szTopic, sizeof(szTopic), "*.*.*", g_szMQTargetOrganizationCode);
        /*
         *  no-ack : 0
         *  no-local : 0
         *  no-wait : 0
         */
        snprintf(szTopic, sizeof(szTopic), "%s.#", g_szMQOrganizationCode);
        amqp_queue_bind(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer),
                        amqp_cstring_bytes(g_szMQFileRecvExchangername), amqp_cstring_bytes(szTopic),
                        amqp_empty_table);
        sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error[메시지 수신 바인딩]!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        }
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Unbinding");
    }

    pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);

    // End of Connect Publish MQ

    // Connect Consume MQ
    pthread_mutex_lock(&pThreadInfo->sConsumeMQ.sMQMutex);

    if (pThreadInfo->sConsumeMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sConsumeMQ.sMQConnectionState = amqp_new_connection();
        //}

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sConsumeMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sConsumeMQ.sMQConnectionState);
            if (pThreadInfo->sConsumeMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Message Queue socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
        //}

        nResult = amqp_socket_open(pThreadInfo->sConsumeMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */


        sAMQPReply = amqp_login(pThreadInfo->sConsumeMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Message Queue!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            break;
        case MQ_ERR_LIB_EXCEPTION:
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sConsumeMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sConsumeMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        AMQPReplyProcess(&sAMQPReply, "amqp_channel_open error[Consume채널]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_OPEN_CHANNEL;
    }
    pThreadInfo->sConsumeMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sConsumeMQ.sMQConnectionState,1);

    amqp_basic_qos(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, g_nPrefetchSize, g_nPrefetchCount, 0);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        AMQPReplyProcess(&sAMQPReply, "amqp_basic_qos error[Consume채널]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        return MQ_ERR_OPEN_CHANNEL;
    }

    if ((nTypeFlags & SOCK_CHANNEL_MESSAGE_RECV) != 0)
    {
        // 메시지 수신 응답 큐 -> Consumer로 이동
        snprintf(szStringBuffer, sizeof(szStringBuffer), "%s%s.%02d", g_szMQMessageRecvRespQueuename, g_szMQTargetOrganizationCode, pThreadInfo->nSocketIndex);
        pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);
        if (pQueueDeclareResult->message_count > 0)
        {
            pQueuePurgeResult = amqp_queue_purge(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer));
            WriteLog(LogType_StandardOut, LogLevel_Info, "MQ Purged[Queuename:%s, MessageCount:%d]!", szStringBuffer, pQueuePurgeResult->message_count);
        }
        // Consume
        // 메시지 수신 응답 큐 Consume
        amqp_basic_consume(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), amqp_empty_bytes, 1, 0, 0, amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
    }

    if ((nTypeFlags & SOCK_CHANNEL_MESSAGE_SEND) != 0)
    {
        amqp_exchange_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQMessageSendExchangername),
                              amqp_cstring_bytes("topic"), 0, 1, 0, 0,
                              amqp_empty_table);

        // declare 오류 체크 방법 확인 필요(Example에는 설명 없음, 문서에도 설명 없음)
        // 수신 큐의 socket index는 확인이 불가능 하므로 메인 큐만 생성한다.
        // 수신 큐의 index는 processing thread에서 처리한다.
        snprintf(szStringBuffer, sizeof(szStringBuffer), "%s%s", g_szMQMessageSendQueuename, g_szMQTargetOrganizationCode);
        amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);

        snprintf(szTopic, sizeof(szTopic), "%s.#", g_szMQTargetOrganizationCode);
        amqp_queue_bind(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer),
                        amqp_cstring_bytes(g_szMQMessageSendExchangername), amqp_cstring_bytes(szTopic),
                        amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Unbinding");
        sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error[메시지 송신 바인딩]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        }
        amqp_basic_consume(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), amqp_empty_bytes, 1, 0, 0, amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
        sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            AMQPReplyProcess(&sAMQPReply, "amqp_basic_consume error[메시지 송신 큐]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        }
    }

    if ((nTypeFlags & SOCK_CHANNEL_FILE_SEND) != 0)
    {
        amqp_exchange_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileSendExchangername),
                              amqp_cstring_bytes("topic"), 0, 1, 0, 0,
                              amqp_empty_table);

        snprintf(szStringBuffer, sizeof(szStringBuffer), "%s%s", g_szMQFileSendQueuename, g_szMQTargetOrganizationCode);
        amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);

        snprintf(szTopic, sizeof(szTopic), "%s.*", g_szMQTargetOrganizationCode);
        amqp_queue_bind(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer),
                        amqp_cstring_bytes(g_szMQFileSendExchangername), amqp_cstring_bytes(szTopic),
                        amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Unbinding");
        sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error[메시지 수신 응답 바인딩]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        }
        amqp_basic_consume(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), amqp_empty_bytes, 1, 0, 0, amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            AMQPReplyProcess(&sAMQPReply, "amqp_basic_consume error[메시지 수신 응답]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        }
    }
    pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);


    // End of Connect Consume MQ


    /*
     * Exchange Declare
         char g_szMQMessageSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQMessageRecvExchangername[MAX_STRING_SIZE + 1];
         char g_szMQFileSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQFileRecvExchangername[MAX_STRING_SIZE + 1];
     *
     */


    /*
     * Queue Declare
     */
    /*

    if ((nTypeFlags & SOCK_CHANNEL_MESSAGE_SEND) != 0)
    {
    }

    if ((nTypeFlags & SOCK_CHANNEL_MESSAGE_RECV) != 0)
    {
    }

    if ((nTypeFlags & SOCK_CHANNEL_FILE_SEND) != 0)
    {
        amqp_exchange_declare(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileSendExchangername),
                              amqp_cstring_bytes("topic"), 0, 1, 0, 0,
                              amqp_empty_table);

        amqp_queue_declare(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileSendQueuename), 0, 1, 0, 0, amqp_empty_table);

        snprintf(szTopic, sizeof(szTopic), "%s.*", g_szMQTargetOrganizationCode);
        amqp_queue_bind(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileSendQueuename),
                        amqp_cstring_bytes(g_szMQFileSendExchangername), amqp_cstring_bytes(szTopic),
                        amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Unbinding");
        amqp_basic_consume(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileSendQueuename), amqp_empty_bytes, 1, 0, 0,
                           amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
    }

    if ((nTypeFlags & SOCK_CHANNEL_FILE_RECV) != 0)
    {
        amqp_exchange_declare(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileRecvExchangername),
                              amqp_cstring_bytes("topic"), 0, 1, 0, 0,
                              amqp_empty_table);

        amqp_queue_declare(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileRecvExchangername), 0, 1, 0, 0, amqp_empty_table);
        //snprintf(szTopic, sizeof(szTopic), "*.*.*", g_szMQTargetOrganizationCode);
        *
         *  no-ack : 0
         *  no-local : 0
         *  no-wait : 0
         *
        amqp_queue_bind(pThreadInfo->sMQConnectionState, 1, amqp_cstring_bytes(g_szMQFileSendQueuename),
                        amqp_cstring_bytes(g_szMQFileSendExchangername), amqp_cstring_bytes("*.*"),
                        amqp_empty_table);
        //die_on_amqp_error(amqp_get_rpc_reply(conn), "Unbinding");
    }

    pthread_mutex_unlock(&pThreadInfo->sMQMutex);
    */
    return MQ_ERR_SUCCESS;
}

int ConnectSenderMQ(struct THREAD_INFO *pThreadInfo, int bMessage)
{
    int nResult;
    amqp_rpc_reply_t sAMQPReply;
    //int nChannelNo;
    //int nTypeFlags;
    //char szStringBuffer[MAX_STRING_SIZE + 1];
    //char szTopic[MAX_STRING_SIZE + 1];
    char szReplyQueuename[MAX_STRING_SIZE + 1];
    amqp_queue_declare_ok_t *pQueueDeclareResult;
    amqp_queue_purge_ok_t *pQueuePurgeResult;

    // Connect Publish MQ
    pthread_mutex_lock(&pThreadInfo->sPublishMQ.sMQMutex);

    if (pThreadInfo->sPublishMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sPublishMQ.sMQConnectionState = amqp_new_connection();
        //}
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ connection created!", pThreadInfo->nThreadIndex);

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sPublishMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sPublishMQ.sMQConnectionState);
            if (pThreadInfo->sPublishMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket create error!", pThreadInfo->nThreadIndex);
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Publish MQ socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                pthread_mutex_unlock(&g_sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket created!", pThreadInfo->nThreadIndex);
        //}

        nResult = amqp_socket_open(pThreadInfo->sPublishMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket open error!", pThreadInfo->nThreadIndex);
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            pthread_mutex_unlock(&g_sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket opened!", pThreadInfo->nThreadIndex);

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */

        sAMQPReply = amqp_login(pThreadInfo->sPublishMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Publish MQ!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login success!", pThreadInfo->nThreadIndex);
            break;
        case MQ_ERR_LIB_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sPublishMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sPublishMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sPublishMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sPublishMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ channel open failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
        pthread_mutex_unlock(&g_sMQMutex);
        return MQ_ERR_OPEN_CHANNEL;
    }
    WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ channel opened!", pThreadInfo->nThreadIndex);
    pThreadInfo->sPublishMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sPublishMQ.sMQConnectionState,1);

    // Exchanger를 여기서 할당해도 의미가 없다.
    // GW에서 Exchager를 바인드 하지 않으면 의미가 없다.
    // 송신큐를 할당된 기관을 알 수 없으므로 기관 Gateway에서 생성하는 것으로 변경
    // Send Exchanger만 알면 처리 가능
//    amqp_queue_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQMessageSendQueuename), 0, 1, 0, 0, amqp_empty_table);

    // bind도 필요가 없다. GW에서 처리 한다.
//    snprintf(szTopic, sizeof(szTopic), "%s.#", g_szMQTargetOrganizationCode);
//    amqp_queue_bind(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQMessageSendQueuename), amqp_cstring_bytes(g_szMQMessageSendExchangername), amqp_cstring_bytes(szTopic), amqp_empty_table);

    pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);

    // End of Connect Publish MQ

    // Connect Consume MQ

    pthread_mutex_lock(&pThreadInfo->sConsumeMQ.sMQMutex);

    if (pThreadInfo->sConsumeMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sConsumeMQ.sMQConnectionState = amqp_new_connection();
        //}
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ connection created!", pThreadInfo->nThreadIndex);

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sConsumeMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sConsumeMQ.sMQConnectionState);
            if (pThreadInfo->sConsumeMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket create error!", pThreadInfo->nThreadIndex);
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Consume MQ socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket created!", pThreadInfo->nThreadIndex);
        //}

        nResult = amqp_socket_open(pThreadInfo->sConsumeMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket open error!", pThreadInfo->nThreadIndex);
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket opened!", pThreadInfo->nThreadIndex);

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */

        sAMQPReply = amqp_login(pThreadInfo->sConsumeMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Consume MQ!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login success!", pThreadInfo->nThreadIndex);
            break;
        case MQ_ERR_LIB_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sConsumeMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sConsumeMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel open failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_OPEN_CHANNEL;
    }
    WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel opened!", pThreadInfo->nThreadIndex);
    pThreadInfo->sConsumeMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sConsumeMQ.sMQConnectionState,1);

    amqp_basic_qos(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, g_nPrefetchSize, g_nPrefetchCount, 0);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        AMQPReplyProcess(&sAMQPReply, "amqp_basic_qos error[Consume채널]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        return MQ_ERR_OPEN_CHANNEL;
    }

    // Message의 경우  Sender가 계속 실행되고 있으므로 named queue를 할당하고 파일의 경우 Sender가 한번 전송하고 사라지기 때문에 임시큐를 이용한다.
    // 임시큐의 경우 autodelete를 사용하지 않으면 큐가 계속 남아있어 조회할 방법이 없다.
    if (bMessage)
    {
        snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendRespQueuename, g_szMQTargetOrganizationCode, pThreadInfo->nThreadIndex);
        pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 1, 0, 0, amqp_empty_table);
    }
    else
    {
        szReplyQueuename[0] = '\0';
        pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 0, 0, 1, amqp_empty_table);
        SafeStrNCopy(g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename), pQueueDeclareResult->queue.bytes, pQueueDeclareResult->queue.len);
    }
//    pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 1, 0, 0, amqp_empty_table);
//    if (!bMessage)
//    {
//        SafeStrCopy(g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename), pQueueDeclareResult->queue.bytes, pQueueDeclareResult->queue.len);
//    }
    if (pQueueDeclareResult->message_count > 0)
    {
        pQueuePurgeResult = amqp_queue_purge(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename));
        WriteLog(LogType_StandardOut, LogLevel_Info, "Consume MQ Purged[Queuename:%s, MessageCount:%d]!", szReplyQueuename, pQueuePurgeResult->message_count);
    }
    // 처음에 rabbitmq 샘플에 direct exchanger를 사용할 경우 amq.direct에 topic 등록하라고 되어 잇었지만
    // rabbitmq 샘플 소스에 exchanger 이름 없이 topic에 Queue 이름을 적을 경우 해당 큐에 바로 저장됨
    //amqp_queue_bind(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(szReplyQueuename), amqp_empty_table);
//    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
//    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
//    {
//        nResult = AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
//        //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is failed!", pThreadInfo->nThreadIndex);
//    }

    amqp_basic_consume(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), amqp_empty_bytes, 1, 0, 0, amqp_empty_table);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        nResult = AMQPReplyProcess(&sAMQPReply, "amqp_basic_consume error[송신 메시지 응답]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_SUCCESS;
    }
    else
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is succeeded!", pThreadInfo->nThreadIndex);
    }

    pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);

    // End of Connect Consume MQ


    /*
     * 아래 두 변수를 이용하여 사용하는 큐를 확정하고 이를 초기화 해야 한다.
     * Message 수신을 사용하면 수신 큐, 수신 응답 큐 바인딩
     * Message 송신을 사용하면 송신 큐, 송신
    int g_nChannelCount = 0;
    int g_nChannelTypeFlags[SOCK_CONF_MAX_CHANNEL_LIMIT];

    SOCK_CHANNEL_MESSAGE_SEND   = 0x01,
    SOCK_CHANNEL_MESSAGE_RECV   = 0x02,
    SOCK_CHANNEL_FILE_SEND      = 0x04,
    SOCK_CHANNEL_FILE_RECV      = 0x08,
    */


    /*
     * Exchange Declare
         char g_szMQMessageSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQMessageRecvExchangername[MAX_STRING_SIZE + 1];
         char g_szMQFileSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQFileRecvExchangername[MAX_STRING_SIZE + 1];
     *
     */

    /*
     * Queue Declare
     */

    return MQ_ERR_SUCCESS;
}
#ifdef _ETCS_ //2개의 Exchange를 사용하기 위함
int ConnectSenderMultiMQ(struct THREAD_INFO *pThreadInfo, int bMessage, int type, int rule)
{
    int nResult;
    amqp_rpc_reply_t sAMQPReply;
    int nChannelNo;
    int nTypeFlags;
    char szStringBuffer[MAX_STRING_SIZE + 1];
    char szTopic[MAX_STRING_SIZE + 1];
    char szReplyQueuename[MAX_STRING_SIZE + 1];
    amqp_queue_declare_ok_t *pQueueDeclareResult;
    amqp_queue_purge_ok_t *pQueuePurgeResult;

    // Connect Publish MQ
    pthread_mutex_lock(&pThreadInfo->sPublishMQ.sMQMutex);

    if (pThreadInfo->sPublishMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sPublishMQ.sMQConnectionState = amqp_new_connection();
        //}
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ connection created!", pThreadInfo->nThreadIndex);

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sPublishMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sPublishMQ.sMQConnectionState);
            if (pThreadInfo->sPublishMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket create error!", pThreadInfo->nThreadIndex);
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Publish MQ socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                pthread_mutex_unlock(&g_sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket created!", pThreadInfo->nThreadIndex);
        //}

        nResult = amqp_socket_open(pThreadInfo->sPublishMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket open error!", pThreadInfo->nThreadIndex);
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            pthread_mutex_unlock(&g_sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket opened!", pThreadInfo->nThreadIndex);

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */

        sAMQPReply = amqp_login(pThreadInfo->sPublishMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Publish MQ!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login success!", pThreadInfo->nThreadIndex);
            break;
        case MQ_ERR_LIB_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sPublishMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sPublishMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sPublishMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sPublishMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ channel open failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
        pthread_mutex_unlock(&g_sMQMutex);
        return MQ_ERR_OPEN_CHANNEL;
    }
    WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ channel opened!", pThreadInfo->nThreadIndex);
    pThreadInfo->sPublishMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sPublishMQ.sMQConnectionState,1);

    // Exchanger를 여기서 할당해도 의미가 없다.
    // GW에서 Exchager를 바인드 하지 않으면 의미가 없다.
    // 송신큐를 할당된 기관을 알 수 없으므로 기관 Gateway에서 생성하는 것으로 변경
    // Send Exchanger만 알면 처리 가능
//    amqp_queue_declare(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQMessageSendQueuename), 0, 1, 0, 0, amqp_empty_table);

    // bind도 필요가 없다. GW에서 처리 한다.
//    snprintf(szTopic, sizeof(szTopic), "%s.#", g_szMQTargetOrganizationCode);
//    amqp_queue_bind(pThreadInfo->sPublishMQ.sMQConnectionState, 1, amqp_cstring_bytes(g_szMQMessageSendQueuename), amqp_cstring_bytes(g_szMQMessageSendExchangername), amqp_cstring_bytes(szTopic), amqp_empty_table);

    pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);

    // End of Connect Publish MQ

    // Connect Consume MQ

    pthread_mutex_lock(&pThreadInfo->sConsumeMQ.sMQMutex);

    if (pThreadInfo->sConsumeMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sConsumeMQ.sMQConnectionState = amqp_new_connection();
        //}
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ connection created!", pThreadInfo->nThreadIndex);

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sConsumeMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sConsumeMQ.sMQConnectionState);
            if (pThreadInfo->sConsumeMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket create error!", pThreadInfo->nThreadIndex);
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Consume MQ socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket created!", pThreadInfo->nThreadIndex);
        //}

        nResult = amqp_socket_open(pThreadInfo->sConsumeMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket open error!", pThreadInfo->nThreadIndex);
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket opened!", pThreadInfo->nThreadIndex);

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */


        sAMQPReply = amqp_login(pThreadInfo->sConsumeMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Consume MQ!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login success!", pThreadInfo->nThreadIndex);
            break;
        case MQ_ERR_LIB_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sConsumeMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sConsumeMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel open failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_OPEN_CHANNEL;
    }
    WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel opened!", pThreadInfo->nThreadIndex);
    pThreadInfo->sConsumeMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sConsumeMQ.sMQConnectionState,1);

    amqp_basic_qos(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, g_nPrefetchSize, g_nPrefetchCount, 0);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        AMQPReplyProcess(&sAMQPReply, "amqp_basic_qos error[Consume채널]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        return MQ_ERR_OPEN_CHANNEL;
    }

    // Message의 경우  Sender가 계속 실행되고 있으므로 named queue를 할당하고 파일의 경우 Sender가 한번 전송하고 사라지기 때문에 임시큐를 이용한다.
    // 임시큐의 경우 autodelete를 사용하지 않으면 큐가 계속 남아있어 조회할 방법이 없다.
    if (bMessage)
    {
        switch(type)
        {
            case 0:
            case 3:
                snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendRespQueuename, g_szMQOrganizationCode, pThreadInfo->nThreadIndex);
                pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 1, 0, 0, amqp_empty_table);
                break;
            case 1:
            case 2:
                if(rule == 0)
                    snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendRealQueuename, g_szMQOrganizationCode, pThreadInfo->nThreadIndex);
                else
                    snprintf(szReplyQueuename, sizeof(szReplyQueuename), "%s%s.%02d", g_szMQMessageSendHqctQueuename, g_szMQOrganizationCode, pThreadInfo->nThreadIndex);
                pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 1, 0, 0, amqp_empty_table);
                break;
            default:
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel unknown type!", pThreadInfo->nThreadIndex);
                pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
                return MQ_ERR_UNKNOWN;
        }
    }
    else
    {
        szReplyQueuename[0] = '\0';
        pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 0, 0, 1, amqp_empty_table);
        SafeStrNCopy(g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename), pQueueDeclareResult->queue.bytes, pQueueDeclareResult->queue.len);
    }
//    pQueueDeclareResult = amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), 0, 1, 0, 0, amqp_empty_table);
//    if (!bMessage)
//    {
//        SafeStrCopy(g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename), pQueueDeclareResult->queue.bytes, pQueueDeclareResult->queue.len);
//    }
    if (pQueueDeclareResult->message_count > 0)
    {
        pQueuePurgeResult = amqp_queue_purge(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename));
        WriteLog(LogType_StandardOut, LogLevel_Info, "Consume MQ Purged[Queuename:%s, MessageCount:%d]!", szReplyQueuename, pQueuePurgeResult->message_count);
    }
    // 처음에 rabbitmq 샘플에 direct exchanger를 사용할 경우 amq.direct에 topic 등록하라고 되어 잇었지만
    // rabbitmq 샘플 소스에 exchanger 이름 없이 topic에 Queue 이름을 적을 경우 해당 큐에 바로 저장됨
    //amqp_queue_bind(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(szReplyQueuename), amqp_empty_table);
//    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
//    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
//    {
//        nResult = AMQPReplyProcess(&sAMQPReply, "amqp_queue_bind error!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
//        //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is failed!", pThreadInfo->nThreadIndex);
//    }

    amqp_basic_consume(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szReplyQueuename), amqp_empty_bytes, 1, 0, 0, amqp_empty_table);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        nResult = AMQPReplyProcess(&sAMQPReply, "amqp_basic_consume error[송신 메시지 응답]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_SUCCESS;
    }
    else
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is succeeded!", pThreadInfo->nThreadIndex);
    }

    pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);

    // End of Connect Consume MQ

    /*
     * 아래 두 변수를 이용하여 사용하는 큐를 확정하고 이를 초기화 해야 한다.
     * Message 수신을 사용하면 수신 큐, 수신 응답 큐 바인딩
     * Message 송신을 사용하면 송신 큐, 송신
    int g_nChannelCount = 0;
    int g_nChannelTypeFlags[SOCK_CONF_MAX_CHANNEL_LIMIT];

    SOCK_CHANNEL_MESSAGE_SEND   = 0x01,
    SOCK_CHANNEL_MESSAGE_RECV   = 0x02,
    SOCK_CHANNEL_FILE_SEND      = 0x04,
    SOCK_CHANNEL_FILE_RECV      = 0x08,
    */

    /*
     * Exchange Declare
         char g_szMQMessageSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQMessageRecvExchangername[MAX_STRING_SIZE + 1];
         char g_szMQFileSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQFileRecvExchangername[MAX_STRING_SIZE + 1];
     *
     */

    /*
     * Queue Declare
     */

    return MQ_ERR_SUCCESS;
}
#endif
int ConnectProcessorMQ(struct THREAD_INFO *pThreadInfo, int bMessage)
{
    int nResult;
    amqp_rpc_reply_t sAMQPReply;
    //int nChannelNo;
    //int nTypeFlags;
    char szStringBuffer[MAX_STRING_SIZE + 1];
    //char szTopic[MAX_STRING_SIZE + 1];

    // Connect Publish MQ

    if (bMessage == true)
    {
        pthread_mutex_lock(&pThreadInfo->sPublishMQ.sMQMutex);

        if (pThreadInfo->sPublishMQ.bMQConnection == false)
        {

            //if (pThreadInfo->sMQConnectionState == NULL)
            //{
                pThreadInfo->sPublishMQ.sMQConnectionState = amqp_new_connection();
            //}
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ connection created!", pThreadInfo->nThreadIndex);

            //if (pThreadInfo->psMQSocket == NULL)
            //{
                pThreadInfo->sPublishMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sPublishMQ.sMQConnectionState);
                if (pThreadInfo->sPublishMQ.psMQSocket == NULL)
                {
                    // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                    WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket create error!", pThreadInfo->nThreadIndex);
                    WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Publish MQ socket in amqp_tcp_socket_new function!!!");
                    pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                    return MQ_ERR_CREATE_SOCKET;
                }
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket created!", pThreadInfo->nThreadIndex);
            //}

            nResult = amqp_socket_open(pThreadInfo->sPublishMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
            if (nResult)
            {
                // TODO : MQ 소켓 연결 오류 처리
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket open error!", pThreadInfo->nThreadIndex);
                pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                return MQ_ERR_CONNECT_SOCKET;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ socket opened!", pThreadInfo->nThreadIndex);

            /*
             *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                                   "guest", "guest"),
                        "Logging in");
             */

            sAMQPReply = amqp_login(pThreadInfo->sPublishMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
            //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
            /*
         *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
         *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
         *    from the broker when logging in will be represented by the broker closing
         *    the socket. In this case r.library_error will be set to
         *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
         *    error conditions including: invalid vhost, authentication failure.
         *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
         *    exception:
         *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
         *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
         *      of the exception. The client should amqp_send_method() a
         *      amqp_channel_close_ok_t. The channel must be re-opened before it
         *      can be used again. Any resources associated with the channel
         *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
         *      and must be recreated before attempting to use them again.
         *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
         *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
         *      details of the exception. The client amqp_send_method() a
         *      amqp_connection_close_ok_t and disconnect from the broker.
             */
            nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Publish MQ!", &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
            switch (nResult)
            {
            case MQ_ERR_SUCCESS:
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login success!", pThreadInfo->nThreadIndex);
                break;
            case MQ_ERR_LIB_EXCEPTION:
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login failed!", pThreadInfo->nThreadIndex);
                pThreadInfo->sPublishMQ.bMQConnection = false;
                pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                return MQ_ERR_LOGIN_LIBRARY;
            case MQ_ERR_SERVER_EXCEPTION:
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ login failed!", pThreadInfo->nThreadIndex);
                pThreadInfo->sPublishMQ.bMQConnection = false;
                pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
                return MQ_ERR_LOGIN_SERVER;
            }
        }
        pThreadInfo->sPublishMQ.bMQConnection = true;

        amqp_channel_open(pThreadInfo->sPublishMQ.sMQConnectionState, 1);
        sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sPublishMQ.sMQConnectionState);
        if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
        {
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ channel open failed!", pThreadInfo->nThreadIndex);
            pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
            return MQ_ERR_OPEN_CHANNEL;
        }
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Publish MQ channel opened!", pThreadInfo->nThreadIndex);
        pThreadInfo->sPublishMQ.bMQChannelOpen = true;
        amqp_confirm_select(pThreadInfo->sPublishMQ.sMQConnectionState,1);

        pthread_mutex_unlock(&pThreadInfo->sPublishMQ.sMQMutex);
    }
    // End of Connect Publish MQ

    // Connect Consume MQ
    pthread_mutex_lock(&pThreadInfo->sConsumeMQ.sMQMutex);

    if (pThreadInfo->sConsumeMQ.bMQConnection == false)
    {

        //if (pThreadInfo->sMQConnectionState == NULL)
        //{
            pThreadInfo->sConsumeMQ.sMQConnectionState = amqp_new_connection();
        //}
            //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ connection created!", pThreadInfo->nThreadIndex);

        //if (pThreadInfo->psMQSocket == NULL)
        //{
            pThreadInfo->sConsumeMQ.psMQSocket = amqp_tcp_socket_new(pThreadInfo->sConsumeMQ.sMQConnectionState);
            if (pThreadInfo->sConsumeMQ.psMQSocket == NULL)
            {
                // TODO : MQ 소켓 생성 오류 처리(로그 남기고 실패 리턴)
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket create error!", pThreadInfo->nThreadIndex);
                WriteLog(LogType_ErrorOut, LogLevel_Error,"Can't create Consume MQ socket in amqp_tcp_socket_new function!!!");
                pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
                return MQ_ERR_CREATE_SOCKET;
            }
            //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket created!", pThreadInfo->nThreadIndex);
        //}

        nResult = amqp_socket_open(pThreadInfo->sConsumeMQ.psMQSocket, g_szMQServerIP, g_nMQServerPort);
        if (nResult)
        {
            // TODO : MQ 소켓 연결 오류 처리
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket open error!", pThreadInfo->nThreadIndex);
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_CONNECT_SOCKET;
        }
        //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ socket opened!", pThreadInfo->nThreadIndex);

        /*
         *   die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
                               "guest", "guest"),
                    "Logging in");
         */

        sAMQPReply = amqp_login(pThreadInfo->sConsumeMQ.sMQConnectionState, g_szMQVirtualHostname, 0, AMQP_DEFAULT_FRAME_SIZE, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        //sAMQPReply = amqp_login(pThreadInfo->sMQConnectionState, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, g_szMQServiceID, g_szMQServicePWD);
        /*
     *  - r.reply_type == AMQP_RESPONSE_NORMAL. Login completed successfully
     *  - r.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION. In most cases errors
     *    from the broker when logging in will be represented by the broker closing
     *    the socket. In this case r.library_error will be set to
     *    AMQP_STATUS_CONNECTION_CLOSED. This error can represent a number of
     *    error conditions including: invalid vhost, authentication failure.
     *  - r.reply_type == AMQP_RESPONSE_SERVER_EXCEPTION. The broker returned an
     *    exception:
     *    - If r.reply.id == AMQP_CHANNEL_CLOSE_METHOD a channel exception
     *      occurred, cast r.reply.decoded to amqp_channel_close_t* to see details
     *      of the exception. The client should amqp_send_method() a
     *      amqp_channel_close_ok_t. The channel must be re-opened before it
     *      can be used again. Any resources associated with the channel
     *      (auto-delete exchanges, auto-delete queues, consumers) are invalid
     *      and must be recreated before attempting to use them again.
     *    - If r.reply.id == AMQP_CONNECTION_CLOSE_METHOD a connection exception
     *      occurred, cast r.reply.decoded to amqp_connection_close_t* to see
     *      details of the exception. The client amqp_send_method() a
     *      amqp_connection_close_ok_t and disconnect from the broker.
         */
        nResult = AMQPReplyProcess(&sAMQPReply, "Can't login Consume MQ!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        switch (nResult)
        {
        case MQ_ERR_SUCCESS:
            //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login success!", pThreadInfo->nThreadIndex);
            break;
        case MQ_ERR_LIB_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_LOGIN_LIBRARY;
        case MQ_ERR_SERVER_EXCEPTION:
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ login failed!", pThreadInfo->nThreadIndex);
            pThreadInfo->sConsumeMQ.bMQConnection = false;
            pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
            return MQ_ERR_LOGIN_SERVER;
        }
    }
    pThreadInfo->sConsumeMQ.bMQConnection = true;

    amqp_channel_open(pThreadInfo->sConsumeMQ.sMQConnectionState, 1);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel open failed!", pThreadInfo->nThreadIndex);
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        return MQ_ERR_OPEN_CHANNEL;
    }
    //WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consume MQ channel opened!", pThreadInfo->nThreadIndex);
    pThreadInfo->sConsumeMQ.bMQChannelOpen = true;
    amqp_confirm_select(pThreadInfo->sConsumeMQ.sMQConnectionState,1);

    amqp_basic_qos(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, g_nPrefetchSize, g_nPrefetchCount, 0);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    if (sAMQPReply.reply_type != AMQP_RESPONSE_NORMAL)
    {
        pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);
        AMQPReplyProcess(&sAMQPReply, "amqp_basic_qos error[Consume채널]!", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
        return MQ_ERR_OPEN_CHANNEL;
    }

    snprintf(szStringBuffer, sizeof(szStringBuffer), "%s%s", g_szMQMessageRecvQueuename, g_szMQOrganizationCode);
    amqp_queue_declare(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), 0, 1, 0, 0, amqp_empty_table);

    amqp_basic_consume(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, amqp_cstring_bytes(szStringBuffer), amqp_empty_bytes, 1, 0, 0, amqp_empty_table);
    sAMQPReply = amqp_get_rpc_reply(pThreadInfo->sConsumeMQ.sMQConnectionState);
    nResult = AMQPReplyProcess(&sAMQPReply, "amqp_basic_consume", &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen);
    if (nResult != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]Consumer registering is failed!", pThreadInfo->nThreadIndex);
    }

    pthread_mutex_unlock(&pThreadInfo->sConsumeMQ.sMQMutex);

    // End of Connect Consume MQ

    /*
     * 아래 두 변수를 이용하여 사용하는 큐를 확정하고 이를 초기화 해야 한다.
     * Message 수신을 사용하면 수신 큐, 수신 응답 큐 바인딩
     * Message 송신을 사용하면 송신 큐, 송신
    int g_nChannelCount = 0;
    int g_nChannelTypeFlags[SOCK_CONF_MAX_CHANNEL_LIMIT];

    SOCK_CHANNEL_MESSAGE_SEND   = 0x01,
    SOCK_CHANNEL_MESSAGE_RECV   = 0x02,
    SOCK_CHANNEL_FILE_SEND      = 0x04,
    SOCK_CHANNEL_FILE_RECV      = 0x08,
    */

    /*
     * Exchange Declare
         char g_szMQMessageSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQMessageRecvExchangername[MAX_STRING_SIZE + 1];
         char g_szMQFileSendQueuename[MAX_STRING_SIZE + 1];
         char g_szMQFileRecvExchangername[MAX_STRING_SIZE + 1];
     *
     */

    /*
     * Queue Declare
     */

    return MQ_ERR_SUCCESS;
}

int CloseMQ(int nIndex, amqp_connection_state_t sMQConnectionState, int *pbMQConnection, int *pbMQChannelOpen, pthread_mutex_t *pMessageQueueMutex)
{

    int nResult;
    //die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS),"Closing channel");
    //die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS),"Closing connection");
    //die_on_error(amqp_destroy_connection(conn), "Ending connection");
    amqp_rpc_reply_t sMQReply;

    pthread_mutex_lock(pMessageQueueMutex);
    sMQReply = amqp_channel_close(sMQConnectionState, 1, AMQP_REPLY_SUCCESS);
    nResult = AMQPReplyProcess(&sMQReply, "Channel Close Error", pbMQConnection, pbMQChannelOpen);
//    if (nResult != MQ_ERR_SUCCESS)
//        return nResult;
    // 오류가 없는 경우에 처리 오류 if문 추가 필요
    *pbMQChannelOpen = false;
    sMQReply = amqp_connection_close(sMQConnectionState, AMQP_REPLY_SUCCESS);
    nResult = AMQPReplyProcess(&sMQReply, "Socket Close Error", pbMQConnection, pbMQChannelOpen);
//    if (nResult != MQ_ERR_SUCCESS)
//        return nResult;
    // 오류가 없는 경우에 처리 오류 if문 추가 필요
    *pbMQConnection = false;
    nResult = amqp_destroy_connection(sMQConnectionState);
    if (nResult != AMQP_STATUS_OK)
    {
        AMQPStatusProcess(nResult, "MQ Destroy Connection");
        pthread_mutex_unlock(pMessageQueueMutex);
        return MQ_ERR_CLOSE;
    }
    pthread_mutex_unlock(pMessageQueueMutex);
    return MQ_ERR_SUCCESS;
}

int GetMessageFromQueue(amqp_connection_state_t sMQConnectionState, char *pExchangername, int nExchangernameLength, char *pData, int nDataLength, char *szCorrelationID, int nCorrelationIDLength, char *szReplyQueuename, int nReplyQueuenameLength, int *bMQConnection, int *bChannelOpen, uint64_t *pnDeliveryTag, struct timeval *pTimeVal)
{
    amqp_envelope_t sMQEnvelope;
    amqp_rpc_reply_t sMQReply;
    amqp_frame_t sMQFrame;
    amqp_message_t sMQMessage;
    int nResult;

    amqp_maybe_release_buffers(sMQConnectionState);
    sMQReply = amqp_consume_message(sMQConnectionState, &sMQEnvelope, pTimeVal, 0);
    if (sMQReply.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && sMQReply.library_error == AMQP_STATUS_TIMEOUT)
    {
        //WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_consume_message error[%s]", amqp_error_string2(sMQReply.library_error));
        return MQ_ERR_TIMEOUT;
    }

    if (AMQP_RESPONSE_NORMAL != sMQReply.reply_type)
    {
        WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_consume_message error[%s]", amqp_error_string2(sMQReply.library_error));
        if (AMQP_RESPONSE_LIBRARY_EXCEPTION == sMQReply.reply_type && AMQP_STATUS_UNEXPECTED_STATE == sMQReply.library_error)
        {
            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_consume_message Enter AMQP_STATUS_UNEXPECTED_STATE Block");
            nResult = amqp_simple_wait_frame(sMQConnectionState, &sMQFrame);
            if (nResult != AMQP_STATUS_OK)
            {
                // 프로그램 종료 상황, 어떻게 처리할지 좀더 고민해보자
                // 1. Reconnnect
                // 2. Thread 종료 및 신규 Thread 동작 등

                WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame error[%s]", amqp_error_string2(nResult));
                return MQ_ERR_UNKNOWN;
            }

            if (AMQP_FRAME_METHOD == sMQFrame.frame_type)
            {
                WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));
                switch (sMQFrame.payload.method.id)
                {
                case AMQP_BASIC_ACK_METHOD:
                    /* if we've turned publisher confirms on, and we've published a
                    * message here is a message being confirmed.
                    */
                    //amqp_basic_ack(sMQConnectionState, 1, sMQEnvelope.delivery_tag, 0);
                    break;
                case AMQP_BASIC_RETURN_METHOD:
                    /* if a published message couldn't be routed and the mandatory
                    * flag was set this is what would be returned. The message then
                    * needs to be read.
                    */

                    {
                        sMQReply = amqp_read_message(sMQConnectionState, sMQFrame.channel, &sMQMessage, 0);
                        if (AMQP_RESPONSE_NORMAL != sMQReply.reply_type)
                        {
                            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_message_data[%s]", sMQMessage.body);
                            return MQ_ERR_MESSAGE_RETURNED;
                        }

                        amqp_destroy_message(&sMQMessage);
                    }

                    break;

                case AMQP_CHANNEL_CLOSE_METHOD:
                    /* a channel.close method happens when a channel exception occurs,
                     * this can happen by publishing to an exchange that doesn't exist
                     * for example.
                     *
                     * In this case you would need to open another channel redeclare
                     * any queues that were declared auto-delete, and restart any
                     * consumers that were attached to the previous channel.
                     */
                    *bChannelOpen = false;
                    return MQ_ERR_CHANNEL_CLOSED;

                case AMQP_CONNECTION_CLOSE_METHOD:
                    /* a connection.close method happens when a connection exception
                     * occurs, this can happen by trying to use a channel that isn't
                     * open for example.
                     *
                     * In this case the whole connection must be restarted.
                     */
                    *bMQConnection = false;
                    return MQ_ERR_CONNECTION_CLOSED;

                default:
                    //fprintf(stderr, "An unexpected method was received %u\n", sMQFrame.payload.method.id);
                    WriteLog(LogType_ErrorOut, LogLevel_Error, "An unexpected method was received %u", sMQFrame.payload.method.id);
                    return MQ_ERR_UNKNOWN;
                }
            }
        }
    }
    else
    {
        SafeMemCopy(pExchangername, nExchangernameLength, sMQEnvelope.exchange.bytes, sMQEnvelope.exchange.len);
        SafeMemCopy(pData, nDataLength, sMQEnvelope.message.body.bytes, sMQEnvelope.message.body.len);
        pData[sMQEnvelope.message.body.len] = '\0';
        SafeMemCopy(szCorrelationID, nCorrelationIDLength, sMQEnvelope.message.properties.correlation_id.bytes, sMQEnvelope.message.properties.correlation_id.len);
        szCorrelationID[sMQEnvelope.message.properties.correlation_id.len] = '\0';
        SafeMemCopy(szReplyQueuename, nReplyQueuenameLength, sMQEnvelope.message.properties.reply_to.bytes, sMQEnvelope.message.properties.reply_to.len);
        szReplyQueuename[sMQEnvelope.message.properties.reply_to.len] = '\0';
        *pnDeliveryTag = sMQEnvelope.delivery_tag;
        amqp_destroy_envelope(&sMQEnvelope);
    }
    // ack 처리를 추후 DB 처리 후 처리하도록 수정 필요
    //amqp_basic_ack(sMQConnectionState, 1, nDeliveryTag, 0);

    return MQ_ERR_SUCCESS;
}

//int PutMessageToQueue(amqp_connection_state_t sMQConnectionState, int nChannelNo, const char *pszExchangename, const char *pszResponseQueuename, const char *pszTopic, struct MESSAGE_INFO *pMessageInfo, const char *pData, int nDataLength)
//int PutMessageToQueue(struct SOCKET_THREAD_INFO *pThreadInfo, int nChannelNo, const char *pszExchangename, const char *pszResponseQueuename, const char *pszTopic, struct MESSAGE_INFO *pMessageInfo, const char *pData, int nDataLength, char *szUUIDString, int nUUIDLength, int bMakeResponseInfo)
int PutMessageToQueue(amqp_connection_state_t sMQConnectionState, int *pbMQConnection, int *pbChannelOpen, int nChannelNo, const char *pszExchangename, const char *pszResponseQueuename, const char *pszTopic, struct MESSAGE_INFO *pMessageInfo, const char *pData, int nDataLength, char *szUUIDString, int nUUIDLength, int bMakeResponseInfo, struct DOUBLELINKEDLISTINFO *pListInfo)
{
//proc_log("MQ 00\n");
//sleep(3);
    int nResult;
    struct MESSAGE_RESPONSE_INFO *pResponseInfo;
    amqp_basic_properties_t sProperties;
    struct timespec sTimeSpec;
    time_t sNow;
    struct tm tNow;
    long nMillisecond;
    struct DOUBLELINKEDLISTNODE *pNode;
    amqp_frame_t sMQFrame;
    amqp_rpc_reply_t sMQReply;
    amqp_message_t sMQMessage;
    char szUUIDStringTemp[UUID_STRING_SIZE + 1];
    char szReturnedMessage[SOCK_BUFFER_SIZE + 1];

    if (bMakeResponseInfo)
    {
WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "IN SafeMemoryAlloc Loot!");

        pResponseInfo = NULL;
        nResult = SafeMemoryAlloc((void **)&pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO));
        //pResponseInfo = SafeAlloc(pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO));
        //pResponseInfo->pHeaderInfo = SafeMemoryAlloc(sizeof(struct MESSAGE_PACKET));
        SafeMemSet(pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO), 0, sizeof(struct MESSAGE_RESPONSE_INFO));
        //SafeMemCopy(pResponseInfo->pHeaderInfo, sizeof(struct MESSAGE_PACKET), &sMessagePacket, sizeof(struct MESSAGE_PACKET));
        pResponseInfo->nMessageLength = pMessageInfo->nMessageLength;
        nResult = SafeMemoryAlloc((void **)&(pResponseInfo->paMessageData), pResponseInfo->nMessageLength);
        //pResponseInfo->paMessageData = SafeAlloc(pResponseInfo->paMessageData, pResponseInfo->nMessageLength);

        SafeMemCopy(pResponseInfo->paMessageData, pResponseInfo->nMessageLength, pMessageInfo->paMessageData, pMessageInfo->nMessageLength);

        //uuid_generate(pResponseInfo->aUUID);
        //uuid_unparse(pResponseInfo->aUUID, szUUIDString);
#ifdef _WIN32
        uuid_parse(szUUIDString, &pResponseInfo->aUUID);
#else
        uuid_parse(szUUIDString, pResponseInfo->aUUID);
#endif
        //szUUIDString[UUID_STRING_SIZE] = '\0';
        //SafeMemCopy(pResponseInfo->aUUID, sizeof(pResponseInfo->aUUID), sUUIDTemp, sizeof(pResponseInfo->aUUID));
        SafeStrCopy(pResponseInfo->szResponseQueuename, sizeof(pResponseInfo->szResponseQueuename), pszResponseQueuename);
    }
    //snprintf(pResponseInfo->szResponseQueuename, sizeof(pResponseInfo->szResponseQueuename), "%s.%02d", g_szMQMessageSendQueuename, pThreadInfo->nSocketIndex);
    //snprintf(szTopic, sizeof(szTopic), "%s.%s%s", sMessagePacket.sHeader.szOrganizationCode, sMessagePacket.sHeader.szWorkType, sMessagePacket.sHeader.szMsgType);
    //SafeStrNCopy(aMessageBody, sizeof(aMessageBody), pMessageInfo->paMessageData + 2, pMessageInfo->nMessageLength - SOCK_PACKET_SIZE);
    sProperties._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG
                        | AMQP_BASIC_CORRELATION_ID_FLAG;

    sProperties.content_type = amqp_cstring_bytes("text/plain");
    sProperties.delivery_mode = 2;
    sProperties.correlation_id = amqp_cstring_bytes(szUUIDString);

    if (pszResponseQueuename!= NULL)
    {
        sProperties._flags = sProperties._flags | AMQP_BASIC_REPLY_TO_FLAG;
        sProperties.reply_to = amqp_cstring_bytes(pszResponseQueuename);
    }

    if (bMakeResponseInfo)
    {
        clock_gettime(CLOCK_REALTIME, &sTimeSpec);
        sNow = sTimeSpec.tv_sec;
        nMillisecond = sTimeSpec.tv_nsec / 1.0e6; // nanoseconds to milliseconds
        localtime_r(&sNow, &tNow);
        SafeMemCopy(&pResponseInfo->sTransferTime, sizeof(pResponseInfo->sTransferTime), &tNow, sizeof(tNow));
        pResponseInfo->nTransferMilliSecond = nMillisecond;
    }
    nResult = amqp_basic_publish(sMQConnectionState, 1, amqp_cstring_bytes(pszExchangename), amqp_cstring_bytes(pszTopic), 1, 0, &sProperties, amqp_cstring_bytes(pData));

    if (nResult != AMQP_STATUS_OK)
    {
        WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_basic_publish error[%s]!", amqp_error_string2(nResult));
        switch (nResult)
        {
        case AMQP_STATUS_OK:
            break;
        case AMQP_STATUS_TIMER_FAILURE:
        case AMQP_STATUS_HEARTBEAT_TIMEOUT:
        case AMQP_STATUS_NO_MEMORY:
        case AMQP_STATUS_TABLE_TOO_BIG:
        case AMQP_STATUS_CONNECTION_CLOSED:
        case AMQP_STATUS_SSL_ERROR:
#if 0
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
#endif
            return MQ_ERR_SSL;
        case AMQP_STATUS_TCP_ERROR:
            // Socket 재연결 필요 여부 확인
#if 0
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
#endif
            return MQ_ERR_TCP;

            /*
        AMQP_STATUS_OK on success, amqp_status_enum value on failure. Note that basic.publish is an async method, the return value from this function only indicates that the message data was successfully transmitted to the broker. It does not indicate failures that occur on the broker, such as publishing to a non-existent exchange. Possible error values:
        AMQP_STATUS_TIMER_FAILURE: system timer facility returned an error the message was not sent.
        AMQP_STATUS_HEARTBEAT_TIMEOUT: connection timed out waiting for a heartbeat from the broker. The message was not sent.
        AMQP_STATUS_NO_MEMORY: memory allocation failed. The message was not sent.
        AMQP_STATUS_TABLE_TOO_BIG: a table in the properties was too large to fit in a single frame. Message was not sent.
        AMQP_STATUS_CONNECTION_CLOSED: the connection was closed.
        AMQP_STATUS_SSL_ERROR: a SSL error occurred.
        AMQP_STATUS_TCP_ERROR: a TCP error occurred. errno or WSAGetLastError() may provide more information
             */

        }
        nResult = amqp_simple_wait_frame(sMQConnectionState, &sMQFrame);
        if (nResult != AMQP_STATUS_OK)
        {
            // 프로그램 종료 상황, 어떻게 처리할지 좀더 고민해보자
            // 1. Reconnnect
            // 2. Thread 종료 및 신규 Thread 동작 등
            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame error[%s]!", amqp_error_string2(nResult));
#if 0
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
#endif
            return MQ_ERR_PUBLISH_FAIL;
        }
//
//        if (AMQP_FRAME_METHOD == sMQFrame.frame_type)
//        {
//            //WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));
//            switch (sMQFrame.payload.method.id)
//            {
//            case AMQP_BASIC_ACK_METHOD:
//                /* if we've turned publisher confirms on, and we've published a
//                * message here is a message being confirmed.
//                */
//
//                //return MQ_ERR_SUCCESS;
//                printf("AMQP_BASIC_ACK_METHOD\n");
//                break;
//            case AMQP_BASIC_NACK_METHOD:
//                /* if we've turned publisher confirms on, and we've published a
//                * message here is a message being confirmed.
//                */
//
//                if (bMakeResponseInfo)
//                {
//                    SafeFree((void **)&(pResponseInfo->paMessageData));
//                    SafeFree((void **)&pResponseInfo);
//                }
//                return MQ_ERR_PUBLISH_FAIL;
//            case AMQP_BASIC_DELIVER_METHOD:
//                break;
//            case AMQP_BASIC_RETURN_METHOD:
//                /* if a published message couldn't be routed and the mandatory
//                * flag was set this is what would be returned. The message then
//                * needs to be read.
//                */
//
//                {
//                    sMQReply = amqp_read_message(sMQConnectionState, sMQFrame.channel, &sMQMessage, 0);
//                    if (AMQP_RESPONSE_NORMAL != sMQReply.reply_type)
//                    {
//                        return MQ_ERR_MESSAGE_RETURNED;
//                    }
//
//                    amqp_destroy_message(&sMQMessage);
//                }
//
//                if (bMakeResponseInfo)
//                {
//                    SafeFree((void **)&(pResponseInfo->paMessageData));
//                    SafeFree((void **)&pResponseInfo);
//                }
//                return MQ_ERR_PUBLISH_FAIL;
//
//            case AMQP_CHANNEL_CLOSE_METHOD:
//                /* a channel.close method happens when a channel exception occurs,
//                 * this can happen by publishing to an exchange that doesn't exist
//                 * for example.
//                 *
//                 * In this case you would need to open another channel redeclare
//                 * any queues that were declared auto-delete, and restart any
//                 * consumers that were attached to the previous channel.
//                 */
//    //            pThreadInfo->bChannelOpen = false;
//                *pbChannelOpen = false;
//                if (bMakeResponseInfo)
//                {
//                    SafeFree((void **)&(pResponseInfo->paMessageData));
//                    SafeFree((void **)&pResponseInfo);
//                }
//                return MQ_ERR_CHANNEL_CLOSED;
//
//            case AMQP_CONNECTION_CLOSE_METHOD:
//                /* a connection.close method happens when a connection exception
//                 * occurs, this can happen by trying to use a channel that isn't
//                 * open for example.
//                 *
//                 * In this case the whole connection must be restarted.
//                 */
//    //            pThreadInfo->bMQConnection = false;
//                *pbMQConnection = false;
//                if (bMakeResponseInfo)
//                {
//                    SafeFree((void **)&(pResponseInfo->paMessageData));
//                    SafeFree((void **)&pResponseInfo);
//                }
//                return MQ_ERR_CONNECTION_CLOSED;
//
//            default:
//                //fprintf(stderr, "An unexpected method was received %u\n", sMQFrame.payload.method.id);
//                WriteLog(LogType_ErrorOut, LogLevel_Error, "An unexpected method was received %u", sMQFrame.payload.method.id);
//                if (bMakeResponseInfo)
//                {
//                    SafeFree((void **)&(pResponseInfo->paMessageData));
//                    SafeFree((void **)&pResponseInfo);
//                }
//                return MQ_ERR_UNKNOWN;
//            }
//        }
    }

    nResult = amqp_simple_wait_frame(sMQConnectionState, &sMQFrame);
    if (nResult != AMQP_STATUS_OK)
    {
        // 프로그램 종료 상황, 어떻게 처리할지 좀더 고민해보자
        // 1. Reconnnect
        // 2. Thread 종료 및 신규 Thread 동작 등
        WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame error[%s]!", amqp_error_string2(nResult));
#if 0
        if (bMakeResponseInfo)
        {
            SafeFree((void **)&(pResponseInfo->paMessageData));
            SafeFree((void **)&pResponseInfo);
        }
#endif
        return MQ_ERR_PUBLISH_FAIL;
    }

    if (AMQP_FRAME_METHOD == sMQFrame.frame_type)
    {
        //WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));
        switch (sMQFrame.payload.method.id)
        {
        case AMQP_BASIC_ACK_METHOD:
            /* if we've turned publisher confirms on, and we've published a
            * message here is a message being confirmed.
            */
            //return MQ_ERR_SUCCESS;
            printf("AMQP_BASIC_ACK_METHOD\n");
            break;
        case AMQP_BASIC_NACK_METHOD:
            /* if we've turned publisher confirms on, and we've published a
            * message here is a message being confirmed.
            */
            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
            return MQ_ERR_PUBLISH_FAIL;
        case AMQP_BASIC_DELIVER_METHOD:
            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));
            break;
        case AMQP_BASIC_RETURN_METHOD:
            /* if a published message couldn't be routed and the mandatory
            * flag was set this is what would be returned. The message then
            * needs to be read.
            */
            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));

            {
                sMQReply = amqp_read_message(sMQConnectionState, sMQFrame.channel, &sMQMessage, 0);
                if (sMQMessage.properties.correlation_id.len != 0)
                {
                    SafeMemCopy(szUUIDStringTemp, sizeof(szUUIDStringTemp), sMQMessage.properties.correlation_id.bytes, sMQMessage.properties.correlation_id.len);
                    szUUIDStringTemp[sMQMessage.properties.correlation_id.len] = '\0';
                    SafeMemCopy(szReturnedMessage, sizeof(szReturnedMessage), sMQMessage.body.bytes, sMQMessage.body.len);
                    szReturnedMessage[sMQMessage.body.len] = '\0';
                    WriteLog(LogType_StandardOut, LogLevel_Info, "RETURNED MESSAGE[CorrerationID:%s,DATA:%s]", szUUIDStringTemp, szReturnedMessage);
                }
                else
                {
                    SafeMemCopy(szReturnedMessage, sizeof(szReturnedMessage), sMQMessage.body.bytes, sMQMessage.body.len);
                    szReturnedMessage[sMQMessage.body.len] = '\0';
                    WriteLog(LogType_StandardOut, LogLevel_Info, "RETURNED MESSAGE[DATA:%s]", szReturnedMessage);
                }
                if (AMQP_RESPONSE_NORMAL != sMQReply.reply_type)
                {
#if 0
                    if (bMakeResponseInfo)
                    {
                        SafeFree((void **)&(pResponseInfo->paMessageData));
                        SafeFree((void **)&pResponseInfo);
                    } 
#endif
                    return MQ_ERR_MESSAGE_RETURNED;
                }
                amqp_destroy_message(&sMQMessage);
            }
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
            // return 발생 후 ack가 바로 발생하는것 같아서 추가해서 확인중
            nResult = amqp_simple_wait_frame(sMQConnectionState, &sMQFrame);
            if (nResult != AMQP_STATUS_OK)
            {
                // 프로그램 종료 상황, 어떻게 처리할지 좀더 고민해보자
                // 1. Reconnnect
                // 2. Thread 종료 및 신규 Thread 동작 등
                WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame error[%s]!", amqp_error_string2(nResult));
                return MQ_ERR_PUBLISH_FAIL;
            }

            return MQ_ERR_PUBLISH_FAIL;

        case AMQP_CHANNEL_CLOSE_METHOD:
            /* a channel.close method happens when a channel exception occurs,
             * this can happen by publishing to an exchange that doesn't exist
             * for example.
             *
             * In this case you would need to open another channel redeclare
             * any queues that were declared auto-delete, and restart any
             * consumers that were attached to the previous channel.
             */
//            pThreadInfo->bChannelOpen = false;
            *pbChannelOpen = false;
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
            return MQ_ERR_CHANNEL_CLOSED;

        case AMQP_CONNECTION_CLOSE_METHOD:
            /* a connection.close method happens when a connection exception
             * occurs, this can happen by trying to use a channel that isn't
             * open for example.
             *
             * In this case the whole connection must be restarted.
             */
//            pThreadInfo->bMQConnection = false;
            WriteTraceLog(LogType_ErrorOut, LogLevel_Trace, __FILE__, __LINE__, __FUNCTION__, "amqp_simple_wait_frame[Method Name:%s]", amqp_method_name(sMQFrame.payload.method.id));
            *pbMQConnection = false;
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
            return MQ_ERR_CONNECTION_CLOSED;

        default:
            //fprintf(stderr, "An unexpected method was received %u\n", sMQFrame.payload.method.id);
            WriteLog(LogType_ErrorOut, LogLevel_Error, "An unexpected method was received %u", sMQFrame.payload.method.id);
            if (bMakeResponseInfo)
            {
                SafeFree((void **)&(pResponseInfo->paMessageData));
                SafeFree((void **)&pResponseInfo);
            }
            return MQ_ERR_UNKNOWN;
        }
    }

    if (bMakeResponseInfo)
    {
        //nResult = DoubleLinkedListAddNodeToHead(&sChannelRecvResponseBuffer[nChannelNo], (void *)pResponseInfo, &pNode);
        nResult = DoubleLinkedListAddNodeToHead(pListInfo, (void *)pResponseInfo, &pNode);
        if (nResult != DS_ERR_SUCCESS)
        {
            // List 추가가 안되는 경우가 어떤 경우가 발생 할지? 에러 로그 남기고 어떻게 처리할까?
        }
#if 0
		SafeFree((void **)&(pResponseInfo->paMessageData));
        SafeFree((void **)&pResponseInfo); 
#endif
    }

    // amqp_cstring_bytes함수는 메모리 복사하는것이 아니라 문자열 길이를 저장하고 기존 문자열의 포인트를 저장하는 Wrapper이므로 amqp_bytes_free를 사용하면 안된다.
    //amqp_bytes_free(sProperties.content_type);
    //amqp_bytes_free(sProperties.correlation_id);
    //amqp_bytes_free(mqExchangename);
    //amqp_bytes_free(mqTopic);
    //amqp_bytes_free(mqMessageBody);
    //amqp_bytes_free(sProperties.reply_to);

    return MQ_ERR_SUCCESS;
}

int PutAckMessageToQueue(amqp_connection_state_t sMQConnectionState, const char *pszCorrelationID, const char *pszResponseQueuename, const char *pData, int nDataLength, int *pbMQConnection, int *pbChannelOpen)
{
    int nResult;
    amqp_basic_properties_t sProperties;
    amqp_frame_t sMQFrame;
    amqp_rpc_reply_t sMQReply;
    amqp_message_t sMQMessage;


    //pResponseInfo = SafeAlloc(pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO));
    //pResponseInfo->pHeaderInfo = SafeMemoryAlloc(sizeof(struct MESSAGE_PACKET));
    //SafeMemCopy(pResponseInfo->pHeaderInfo, sizeof(struct MESSAGE_PACKET), &sMessagePacket, sizeof(struct MESSAGE_PACKET));
    //pResponseInfo->paMessageData = SafeAlloc(pResponseInfo->paMessageData, pResponseInfo->nMessageLength);
    //snprintf(pResponseInfo->szResponseQueuename, sizeof(pResponseInfo->szResponseQueuename), "%s.%02d", g_szMQMessageSendQueuename, pThreadInfo->nSocketIndex);
    //snprintf(szTopic, sizeof(szTopic), "%s.%s%s", sMessagePacket.sHeader.szOrganizationCode, sMessagePacket.sHeader.szWorkType, sMessagePacket.sHeader.szMsgType);
    //SafeStrNCopy(aMessageBody, sizeof(aMessageBody), pMessageInfo->paMessageData + 2, pMessageInfo->nMessageLength - SOCK_PACKET_SIZE);
    sProperties._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |
                   AMQP_BASIC_DELIVERY_MODE_FLAG |
                   AMQP_BASIC_CORRELATION_ID_FLAG;
    sProperties.content_type = amqp_cstring_bytes("text/plain");
    sProperties.delivery_mode = 2;
    //sProperties.reply_to = amqp_cstring_bytes(pszResponseQueuename);
    sProperties.correlation_id = amqp_cstring_bytes(pszCorrelationID);

    nResult = amqp_basic_publish(sMQConnectionState, 1, amqp_cstring_bytes(""), amqp_cstring_bytes(pszResponseQueuename), 0, 0, &sProperties, amqp_cstring_bytes(pData));

    switch (nResult)
    {
    case AMQP_STATUS_OK:
        break;
    case AMQP_STATUS_TIMER_FAILURE:
    case AMQP_STATUS_HEARTBEAT_TIMEOUT:
    case AMQP_STATUS_NO_MEMORY:
    case AMQP_STATUS_TABLE_TOO_BIG:
    case AMQP_STATUS_CONNECTION_CLOSED:
    case AMQP_STATUS_SSL_ERROR:
        WriteLog(LogType_ErrorOut, LogLevel_Error, "amqp_basic_publish error[%s]!", amqp_error_string2(nResult));
        return MQ_ERR_SSL;
    case AMQP_STATUS_TCP_ERROR:
        // Socket 재연결 필요 여부 확인
        WriteLog(LogType_ErrorOut, LogLevel_Error, "amqp_basic_publish error[%s]!",strerror(errno));
        return MQ_ERR_TCP;

        /*
    AMQP_STATUS_OK on success, amqp_status_enum value on failure. Note that basic.publish is an async method, the return value from this function only indicates that the message data was successfully transmitted to the broker. It does not indicate failures that occur on the broker, such as publishing to a non-existent exchange. Possible error values:
    AMQP_STATUS_TIMER_FAILURE: system timer facility returned an error the message was not sent.
    AMQP_STATUS_HEARTBEAT_TIMEOUT: connection timed out waiting for a heartbeat from the broker. The message was not sent.
    AMQP_STATUS_NO_MEMORY: memory allocation failed. The message was not sent.
    AMQP_STATUS_TABLE_TOO_BIG: a table in the properties was too large to fit in a single frame. Message was not sent.
    AMQP_STATUS_CONNECTION_CLOSED: the connection was closed.
    AMQP_STATUS_SSL_ERROR: a SSL error occurred.
    AMQP_STATUS_TCP_ERROR: a TCP error occurred. errno or WSAGetLastError() may provide more information
         */

    }
    if (AMQP_STATUS_OK != amqp_simple_wait_frame(sMQConnectionState, &sMQFrame))
    {
        // 프로그램 종료 상황, 어떻게 처리할지 좀더 고민해보자
        // 1. Reconnnect
        // 2. Thread 종료 및 신규 Thread 동작 등
        return MQ_ERR_PUBLISH_FAIL;
    }

    if (AMQP_FRAME_METHOD == sMQFrame.frame_type)
    {
        switch (sMQFrame.payload.method.id)
        {
        case AMQP_BASIC_ACK_METHOD:
            /* if we've turned publisher confirms on, and we've published a
            * message here is a message being confirmed.
            */

            return MQ_ERR_SUCCESS;
        case AMQP_BASIC_NACK_METHOD:
            /* if we've turned publisher confirms on, and we've published a
            * message here is a message being confirmed.
            */

            return MQ_ERR_PUBLISH_FAIL;
        case AMQP_BASIC_RETURN_METHOD:
            /* if a published message couldn't be routed and the mandatory
            * flag was set this is what would be returned. The message then
            * needs to be read.
            */

            {
                sMQReply = amqp_read_message(sMQConnectionState, sMQFrame.channel, &sMQMessage, 0);
                if (AMQP_RESPONSE_NORMAL != sMQReply.reply_type)
                {
                    return MQ_ERR_MESSAGE_RETURNED;
                }

                amqp_destroy_message(&sMQMessage);
            }

            return MQ_ERR_PUBLISH_FAIL;

        case AMQP_CHANNEL_CLOSE_METHOD:
            /* a channel.close method happens when a channel exception occurs,
             * this can happen by publishing to an exchange that doesn't exist
             * for example.
             *
             * In this case you would need to open another channel redeclare
             * any queues that were declared auto-delete, and restart any
             * consumers that were attached to the previous channel.
             */
            *pbChannelOpen = false;
            return MQ_ERR_CHANNEL_CLOSED;

        case AMQP_CONNECTION_CLOSE_METHOD:
            /* a connection.close method happens when a connection exception
             * occurs, this can happen by trying to use a channel that isn't
             * open for example.
             *
             * In this case the whole connection must be restarted.
             */
            *pbMQConnection = false;
            return MQ_ERR_CONNECTION_CLOSED;

        default:
            //fprintf(stderr, "An unexpected method was received %u\n", sMQFrame.payload.method.id);
            WriteLog(LogType_ErrorOut, LogLevel_Error, "An unexpected method was received %u", sMQFrame.payload.method.id);
            return MQ_ERR_UNKNOWN;
        }
    }

    // amqp_cstring_bytes함수는 메모리 복사하는것이 아니라 문자열 길이를 저장하고 기존 문자열의 포인트를 저장하는 Wrapper이므로 amqp_bytes_free를 사용하면 안된다.
    //amqp_bytes_free(sProperties.content_type);
    //amqp_bytes_free(sProperties.correlation_id);
    //amqp_bytes_free(mqExchangename);
    //amqp_bytes_free(mqTopic);
    //amqp_bytes_free(mqMessageBody);
    //amqp_bytes_free(sProperties.reply_to);

    return MQ_ERR_UNKNOWN;
}

int PutReplyMessageToQueue(amqp_connection_state_t sMQConnectionState, const char *pszCorrelationID, const char *pszResponseQueuename, const char *pData, int nDataLength, int *pbMQConnection, int *pbChannelOpen)
{
    int nResult;
    amqp_basic_properties_t sProperties;
    amqp_frame_t sMQFrame;
    amqp_rpc_reply_t sMQReply;
    amqp_message_t sMQMessage;

    //pResponseInfo = SafeAlloc(pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO));
    //pResponseInfo->pHeaderInfo = SafeMemoryAlloc(sizeof(struct MESSAGE_PACKET));
    //SafeMemCopy(pResponseInfo->pHeaderInfo, sizeof(struct MESSAGE_PACKET), &sMessagePacket, sizeof(struct MESSAGE_PACKET));
    //pResponseInfo->paMessageData = SafeAlloc(pResponseInfo->paMessageData, pResponseInfo->nMessageLength);
    //snprintf(pResponseInfo->szResponseQueuename, sizeof(pResponseInfo->szResponseQueuename), "%s.%02d", g_szMQMessageSendQueuename, pThreadInfo->nSocketIndex);
    //snprintf(szTopic, sizeof(szTopic), "%s.%s%s", sMessagePacket.sHeader.szOrganizationCode, sMessagePacket.sHeader.szWorkType, sMessagePacket.sHeader.szMsgType);
    //SafeStrNCopy(aMessageBody, sizeof(aMessageBody), pMessageInfo->paMessageData + 2, pMessageInfo->nMessageLength - SOCK_PACKET_SIZE);
    sProperties._flags = AMQP_BASIC_CONTENT_TYPE_FLAG |  AMQP_BASIC_CORRELATION_ID_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;// | AMQP_BASIC_REPLY_TO_FLAG;
    sProperties.content_type = amqp_cstring_bytes("text/plain");
    //sProperties.reply_to = amqp_cstring_bytes(pszResponseQueuename);
    sProperties.correlation_id = amqp_cstring_bytes(pszCorrelationID);
    // delivery mode :
    sProperties.delivery_mode = 2;

    nResult = amqp_basic_publish(sMQConnectionState, 1, amqp_cstring_bytes(""), amqp_cstring_bytes(pszResponseQueuename), 1, 0, &sProperties, amqp_cstring_bytes(pData));

    switch (nResult)
    {
    case AMQP_STATUS_OK:
        break;
    case AMQP_STATUS_TIMER_FAILURE:
    case AMQP_STATUS_HEARTBEAT_TIMEOUT:
    case AMQP_STATUS_NO_MEMORY:
    case AMQP_STATUS_TABLE_TOO_BIG:
    case AMQP_STATUS_CONNECTION_CLOSED:
    case AMQP_STATUS_SSL_ERROR:
        WriteLog(LogType_ErrorOut, LogLevel_Error, "amqp_basic_publish error[%s]!", amqp_error_string2(nResult));
        return MQ_ERR_SSL;
    case AMQP_STATUS_TCP_ERROR:
        // Socket 재연결 필요 여부 확인
        WriteLog(LogType_ErrorOut, LogLevel_Error, "amqp_basic_publish error[%s]!",strerror(errno));
        return MQ_ERR_TCP;

        /*
    AMQP_STATUS_OK on success, amqp_status_enum value on failure. Note that basic.publish is an async method, the return value from this function only indicates that the message data was successfully transmitted to the broker. It does not indicate failures that occur on the broker, such as publishing to a non-existent exchange. Possible error values:
    AMQP_STATUS_TIMER_FAILURE: system timer facility returned an error the message was not sent.
    AMQP_STATUS_HEARTBEAT_TIMEOUT: connection timed out waiting for a heartbeat from the broker. The message was not sent.
    AMQP_STATUS_NO_MEMORY: memory allocation failed. The message was not sent.
    AMQP_STATUS_TABLE_TOO_BIG: a table in the properties was too large to fit in a single frame. Message was not sent.
    AMQP_STATUS_CONNECTION_CLOSED: the connection was closed.
    AMQP_STATUS_SSL_ERROR: a SSL error occurred.
    AMQP_STATUS_TCP_ERROR: a TCP error occurred. errno or WSAGetLastError() may provide more information
         */

    }
    if (AMQP_STATUS_OK != amqp_simple_wait_frame(sMQConnectionState, &sMQFrame))
    {
        // 프로그램 종료 상황, 어떻게 처리할지 좀더 고민해보자
        // 1. Reconnnect
        // 2. Thread 종료 및 신규 Thread 동작 등
        return MQ_ERR_PUBLISH_FAIL;
    }

    if (AMQP_FRAME_METHOD == sMQFrame.frame_type)
    {
        switch (sMQFrame.payload.method.id)
        {
        case AMQP_BASIC_ACK_METHOD:
            /* if we've turned publisher confirms on, and we've published a
            * message here is a message being confirmed.
            */

            return MQ_ERR_SUCCESS;
        case AMQP_BASIC_NACK_METHOD:
            /* if we've turned publisher confirms on, and we've published a
            * message here is a message being confirmed.
            */

            return MQ_ERR_PUBLISH_FAIL;
        case AMQP_BASIC_RETURN_METHOD:
            /* if a published message couldn't be routed and the mandatory
            * flag was set this is what would be returned. The message then
            * needs to be read.
            */

            {
                sMQReply = amqp_read_message(sMQConnectionState, sMQFrame.channel, &sMQMessage, 0);
                if (AMQP_RESPONSE_NORMAL != sMQReply.reply_type)
                {
                    return MQ_ERR_MESSAGE_RETURNED;
                }

                amqp_destroy_message(&sMQMessage);
            }

            return MQ_ERR_PUBLISH_FAIL;

        case AMQP_CHANNEL_CLOSE_METHOD:
            /* a channel.close method happens when a channel exception occurs,
             * this can happen by publishing to an exchange that doesn't exist
             * for example.
             *
             * In this case you would need to open another channel redeclare
             * any queues that were declared auto-delete, and restart any
             * consumers that were attached to the previous channel.
             */
            *pbChannelOpen = false;
            return MQ_ERR_CHANNEL_CLOSED;

        case AMQP_CONNECTION_CLOSE_METHOD:
            /* a connection.close method happens when a connection exception
             * occurs, this can happen by trying to use a channel that isn't
             * open for example.
             *
             * In this case the whole connection must be restarted.
             */
            *pbMQConnection = false;
            return MQ_ERR_CONNECTION_CLOSED;

        default:
            //fprintf(stderr, "An unexpected method was received %u\n", sMQFrame.payload.method.id);
            WriteLog(LogType_ErrorOut, LogLevel_Error, "An unexpected method was received %u", sMQFrame.payload.method.id);
            return MQ_ERR_UNKNOWN;
        }
    }

    // amqp_cstring_bytes함수는 메모리 복사하는것이 아니라 문자열 길이를 저장하고 기존 문자열의 포인트를 저장하는 Wrapper이므로 amqp_bytes_free를 사용하면 안된다.
    //amqp_bytes_free(sProperties.content_type);
    //amqp_bytes_free(sProperties.correlation_id);
    //amqp_bytes_free(mqExchangename);
    //amqp_bytes_free(mqTopic);
    //amqp_bytes_free(mqMessageBody);
    //amqp_bytes_free(sProperties.reply_to);

    return MQ_ERR_UNKNOWN;
}

int ProcessRecvQueueResponse(struct SOCKET_THREAD_INFO *pThreadInfo, int nChannelNo, const char *aMessageBody, int nMessageBodySize, const char *szUUIDString, int nUUIDStringSize)
{
    int nResult;
    int nStringCmpResult;
    long nTimeDiff;
    //int nMessageBodyDataSize;
    struct DOUBLELINKEDLISTNODE *pNode;
    struct MESSAGE_RESPONSE_INFO *pMessageResponseInfo;
    struct MESSAGE_INFO *pMessageInfo;
    struct timespec sTimeSpec;
    time_t sNow;
    struct tm pNow;
    int nMillisecond;
    //int nPacketLength;
    //char aFrameBuffer[SOCK_BUFFER_SIZE + 1];
    //char aPacketBuffer[SOCK_BUFFER_SIZE + 1];
    char szUUIDStringTemp[UUID_STRING_SIZE + 1];

    pNode = NULL;
    nResult = DoubleLinkedListMoveToHead(&pThreadInfo->sChannelRecvResponseBuffer[nChannelNo], &pNode);
    pMessageInfo = NULL;
    pMessageResponseInfo = NULL;
    while (nResult == DS_ERR_SUCCESS)
    {
        pMessageResponseInfo = (struct MESSAGE_RESPONSE_INFO *)pNode->pNodeData;
#ifdef _WIN32
        uuid_unparse(&pMessageResponseInfo->aUUID, szUUIDStringTemp);
#else
        uuid_unparse(pMessageResponseInfo->aUUID, szUUIDStringTemp);
#endif
        nResult = SafeStrCmp(szUUIDStringTemp, sizeof(szUUIDStringTemp), szUUIDString, &nStringCmpResult);
        if (nResult == SEC_ERR_SUCCESS && nStringCmpResult == 0)
        {
            WriteCommLog(LogType_CommOut, LogLevel_Info, pThreadInfo->nSocket, nChannelNo, 'R', "%-*.*s,%s", UUID_STRING_SIZE, UUID_STRING_SIZE, szUUIDStringTemp, aMessageBody);
            clock_gettime(CLOCK_REALTIME, &sTimeSpec);
            sNow = sTimeSpec.tv_sec;
            nMillisecond = sTimeSpec.tv_nsec / 1.0e6; // nanoseconds to milliseconds
            localtime_r(&sNow, &pNow);

            nTimeDiff = TimeDiff(&pMessageResponseInfo->sTransferTime, pMessageResponseInfo->nTransferMilliSecond, &pNow, nMillisecond);

            // 타임아웃 시간이 지났으면 무시
            if (nTimeDiff >= g_nErrorTimeout || pMessageResponseInfo->bResponse == true)
            {
                WriteLog(LogType_StandardOut, LogLevel_Info, "TIMEOUT is already sended[CorrelationID:%s,MSG:%s]", szUUIDString, aMessageBody);
                pMessageResponseInfo->bTimeout = true;
                pMessageResponseInfo->bResponseRecv = true;
                if (!g_bBypassMode)
                {
                    amqp_basic_nack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, pMessageResponseInfo->nDeliveryTag, 0, 1);
                }
            }
            else
            {
                // 타임아웃 시간 이내이면 전송
//                SafeMemCopy(aFrameBuffer, sizeof(aFrameBuffer), aMessageBody, nMessageBodySize);
//                nMessageBodyDataSize = strnlen_s(aFrameBuffer, nMessageBodySize);
//                nPacketLength = sizeof(aPacketBuffer);
//                MakePacket(nChannelNo, aFrameBuffer, nMessageBodyDataSize, aPacketBuffer, &nPacketLength);
//                nResult = SafeMemoryAlloc((void *)&pMessageInfo, sizeof(struct MESSAGE_INFO));
//                SafeMemSet(pMessageInfo, sizeof (struct MESSAGE_INFO), 0, sizeof(struct MESSAGE_INFO));
//                nResult = SafeMemoryAlloc((void *)&pMessageInfo->paMessageData, nPacketLength + 1);
//                SafeMemSet(pMessageInfo->paMessageData, nPacketLength + 1, 0, nPacketLength + 1);
//                SafeMemCopy(pMessageInfo->paMessageData, nPacketLength, aPacketBuffer, nPacketLength);
//                pMessageInfo->nMessageLength = nPacketLength;
//                pNode = NULL;
//                nResult = DoubleLinkedListAddNodeToHead(&pThreadInfo->sChannelSendFrameBuffer[nChannelNo], pMessageInfo, &pNode);
//              int SendPacket(struct SOCKET_THREAD_INFO *pThreadInfo, int nChannelNo, uuid_t *pUUID, const char *aMessageBody, int nMessageBodySize)
                //const char *aMessageBody, int nMessageBodySize, const char *szUUIDString, int nUUIDStringSize)

                if (g_bBypassMode)
                {
                    nResult = SendPacket(pThreadInfo, nChannelNo, &pMessageResponseInfo->aUUID, aMessageBody, nMessageBodySize);
                }
                else
                {
                    amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, pMessageResponseInfo->nDeliveryTag, 0);
                }

                pMessageResponseInfo->bResponse = true;
                pMessageResponseInfo->bResponse = true;
                pMessageResponseInfo->bResponseRecv = true;
            }

            return MQ_ERR_SUCCESS;
        }
        nResult = DoubleLinkedListMoveToNext(&pThreadInfo->sChannelRecvResponseBuffer[nChannelNo], &pNode);
    }

    WriteTraceLog(LogType_StandardOut, LogLevel_Info, __FILE__, __LINE__, __FUNCTION__, "Can't find original recv message[CorrelationID:%s,MSG:%s]", szUUIDString, aMessageBody);
    return MQ_ERR_CANT_FIND_ORIG_MSG;
}



int QSendMessage(struct SOCKET_THREAD_INFO *pThreadInfo, int nChannelNo, const char *aMessageBody, int nMessageBodySize, const char *szUUIDString, int nUUIDStringSize, const char *szResponseQueuename, int nResponseQueuenameSize, uint64_t nDeliveryTag)
{
    int nPacketLength;
    char aPacketBuffer[SOCK_BUFFER_SIZE + 1];
    struct MESSAGE_INFO *pMessageInfo;
    struct MESSAGE_RESPONSE_INFO *pResponseInfo;
    struct DOUBLELINKEDLISTNODE *pNode;
    struct timespec sTimeSpec;
    time_t sNow;
    int nResult;

    pMessageInfo = NULL;
    pResponseInfo = NULL;
    pNode = NULL;
    nPacketLength = sizeof(aPacketBuffer);
    MakePacket(nChannelNo, aMessageBody, nMessageBodySize, aPacketBuffer, &nPacketLength);
    nResult = SafeMemoryAlloc((void *)&pMessageInfo, sizeof(struct MESSAGE_INFO));
    SafeMemSet(pMessageInfo, sizeof(struct MESSAGE_INFO), 0, sizeof(struct MESSAGE_INFO));
    nResult = SafeMemoryAlloc((void *)&pMessageInfo->paMessageData, nPacketLength + 1);
    memcpy_s(pMessageInfo->paMessageData, nPacketLength, aPacketBuffer, nPacketLength);
    pMessageInfo->paMessageData[nPacketLength] = '\0';
    pMessageInfo->nMessageLength = nPacketLength;
    pNode = NULL;
    nResult = DoubleLinkedListAddNodeToHead(&pThreadInfo->sChannelSendFrameBuffer[nChannelNo], pMessageInfo, &pNode);

    // Insert to ResponseQueue
    nResult = SafeMemoryAlloc((void **)&pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO));
    SafeMemSet(pResponseInfo, sizeof(struct MESSAGE_RESPONSE_INFO), 0, sizeof(struct MESSAGE_RESPONSE_INFO));
    nResult = SafeMemoryAlloc((void **)&pResponseInfo->paMessageData, nMessageBodySize + 1);
    SafeMemSet(pResponseInfo->paMessageData, nMessageBodySize + 1, 0, nPacketLength);
    pResponseInfo->nMessageLength = nMessageBodySize + 1;
    SafeMemCopy(pResponseInfo->paMessageData, nMessageBodySize, aMessageBody, nMessageBodySize);
    pResponseInfo->paMessageData[nMessageBodySize] = '\0';
    pResponseInfo->nDeliveryTag = nDeliveryTag;

#ifdef _WIN32
    uuid_parse(szUUIDString, &pResponseInfo->aUUID);
#else
    uuid_parse(szUUIDString, pResponseInfo->aUUID);
#endif
    SafeStrNCopy(pResponseInfo->szResponseQueuename, sizeof(pResponseInfo->szResponseQueuename), szResponseQueuename, nResponseQueuenameSize);


    clock_gettime(CLOCK_REALTIME, &sTimeSpec);
    sNow = sTimeSpec.tv_sec;
    pResponseInfo->nTransferMilliSecond = sTimeSpec.tv_nsec / 1.0e6; // nanoseconds to milliseconds
    localtime_r(&sNow, &pResponseInfo->sTransferTime);
    pNode = NULL;
    nResult = DoubleLinkedListAddNodeToHead(&pThreadInfo->sChannelSendResponseBuffer[nChannelNo], pResponseInfo, &pNode);

    return MQ_ERR_SUCCESS;
}

int MQReadCheck(amqp_connection_state_t sMQConnectionState)
{
    int nSocket;
    fd_set sReadFDSet;
    struct timeval sTimeout;
    int nResult;

    nSocket = amqp_get_sockfd(sMQConnectionState);
    FD_ZERO(&sReadFDSet);
    FD_SET(nSocket, &sReadFDSet);

    sTimeout.tv_sec = 0;
    sTimeout.tv_usec = 100;

    nResult = select(nSocket + 1, &sReadFDSet, NULL, NULL, &sTimeout);
    if (nResult > 0 && FD_ISSET(nSocket, &sReadFDSet))
    {
        return true;
    }

    return false;
}

int ProcessRecvSocketResponse(struct SOCKET_THREAD_INFO *pThreadInfo, int nChannelNo, const char *aMessageBody, int nMessageBodySize, char *szUUIDString, int nUUIDStringSize, int bSuccess)
{
    int nResult;
    int nStringCmpResult;
    long nTimeDiff;
    struct DOUBLELINKEDLISTNODE *pNode;
    struct MESSAGE_RESPONSE_INFO *pMessageResponseInfo;
    struct MESSAGE_INFO *pMessageInfo;
    struct timespec sTimeSpec;
    time_t sNow;
    struct tm pNow;
    int nMillisecond;
    int nPacketLength;
    char aFrameBuffer[SOCK_BUFFER_SIZE + 1];
    char aPacketBuffer[SOCK_BUFFER_SIZE + 1];
    //char szUUIDStringTemp[UUID_STRING_SIZE + 1];

    pNode = NULL;
    nResult = DoubleLinkedListMoveToHead(&pThreadInfo->sChannelSendResponseBuffer[nChannelNo], &pNode);

    while (nResult == DS_ERR_SUCCESS)
    {
        pMessageResponseInfo = (struct MESSAGE_RESPONSE_INFO *)pNode->pNodeData;
#ifdef _WIN32
        uuid_unparse(&pMessageResponseInfo->aUUID, szUUIDString);
#else
        uuid_unparse(pMessageResponseInfo->aUUID, szUUIDString);
#endif
        nResult = SafeMemCmp(pMessageResponseInfo->paMessageData + g_nMessageUniqueueKeyPos, pMessageResponseInfo->nMessageLength - g_nMessageUniqueueKeyPos, aMessageBody + g_nMessageUniqueueKeyPos, g_nMessageUniqueueKeyLen, &nStringCmpResult);
        if (nResult == SEC_ERR_SUCCESS && nStringCmpResult == 0)
        {
            clock_gettime(CLOCK_REALTIME, &sTimeSpec);
            sNow = sTimeSpec.tv_sec;
            nMillisecond = sTimeSpec.tv_nsec / 1.0e6; // nanoseconds to milliseconds
            localtime_r(&sNow, &pNow);

            nTimeDiff = TimeDiff(&pMessageResponseInfo->sTransferTime, pMessageResponseInfo->nTransferMilliSecond, &pNow, nMillisecond);

            // 타임아웃 시간이 지났으면 무시 : 응답결과 수신 시간이 지난 후 도착한 경우 처리 로직 추가 여부 확인
            if (nTimeDiff >= g_nErrorTimeout || pMessageResponseInfo->bResponse == true)
            {
                WriteLog(LogType_StandardOut, LogLevel_Info, "TIMEOUT is already sended[CorrelationID:%s,MSG:%s]", szUUIDString, aMessageBody);
                pMessageResponseInfo->bTimeout = true;
                pMessageResponseInfo->bResponseRecv = true;
                if (!g_bBypassMode)
                {
                    amqp_basic_nack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, pMessageResponseInfo->nDeliveryTag, 0, 1);
                }
            }
            else
            {
                // 타임아웃 시간 이내이면 전송
                SafeMemCopy(aFrameBuffer, sizeof(aFrameBuffer), aMessageBody, nMessageBodySize);
                nPacketLength = sizeof(aPacketBuffer);
                MakePacket(nChannelNo, aFrameBuffer, nMessageBodySize, aPacketBuffer, &nPacketLength);
                pMessageInfo = NULL;
                nResult = SafeMemoryAlloc((void *)&pMessageInfo, sizeof(struct MESSAGE_INFO));
                SafeMemSet(pMessageInfo, sizeof (struct MESSAGE_INFO), 0, sizeof(struct MESSAGE_INFO));
                pMessageInfo->paMessageData = NULL;
                nResult = SafeMemoryAlloc((void *)&pMessageInfo->paMessageData, nPacketLength);
                SafeMemCopy(pMessageInfo->paMessageData, nPacketLength, aPacketBuffer, nPacketLength);
                pMessageInfo->nMessageLength = nPacketLength;
                pNode = NULL;
                if (g_bBypassMode)
                {
                    //nResult = DoubleLinkedListAddNodeToHead(&pThreadInfo->sChannelSendFrameBuffer[nChannelNo], pMessageInfo, &pNode);
                    //int PutMessageToQueue(struct SOCKET_THREAD_INFO *pThreadInfo, int nChannelNo, const char *pszExchangename, const char *pszResponseQueuename, const char *pszTopic, struct MESSAGE_INFO *pMessageInfo, const char *pData, int nDataLength, char *szUUIDString, int nUUIDLength)
                    // RabbitMQ amq.direct를 사용하지 않을 경우 Exchanger 이름이 없고 reply_to에 대상 큐 이름이 명시되어야 해당 큐로 전송된다.
                    // Topic으로 전송된다고 알고 있었으나 Topic에 큐 이름이 있어도 전송되지 않는다.
                    nResult = PutMessageToQueue(pThreadInfo->sPublishMQ.sMQConnectionState, &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen, nChannelNo, "", pMessageResponseInfo->szResponseQueuename, pMessageResponseInfo->szResponseQueuename, pMessageInfo, aMessageBody, pMessageInfo->nMessageLength - SOCK_PACKET_SIZE, szUUIDString, sizeof(szUUIDString), false, &pThreadInfo->sChannelRecvResponseBuffer[nChannelNo]);
                    //nResult = PutMessageToQueue(pThreadInfo->sPublishMQ.sMQConnectionState, &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen, nChannelNo, "", pMessageResponseInfo->szResponseQueuename, pMessageResponseInfo->szResponseQueuename, pMessageInfo, aMessageBody, pMessageInfo->nMessageLength - SOCK_PACKET_SIZE, szUUIDString, sizeof(szUUIDString), false, &pThreadInfo->sChannelRecvResponseBuffer[nChannelNo]);
                    WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nSocket, 'P', "", pMessageResponseInfo->szResponseQueuename, szUUIDString, pMessageInfo->nMessageLength - SOCK_PACKET_SIZE, "%s", aMessageBody);
                }
                else
                {
                    if (bSuccess)
                    {
                        amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, pMessageResponseInfo->nDeliveryTag, 0);
                    }
                    else
                    {
                        amqp_basic_nack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, pMessageResponseInfo->nDeliveryTag, 0, 1);
                    }
                }
                pMessageResponseInfo->bResponse = true;
                pMessageResponseInfo->bResponseRecv = true;
            }

            return MQ_ERR_SUCCESS;
        }
        nResult = DoubleLinkedListMoveToNext(&pThreadInfo->sChannelSendResponseBuffer[nChannelNo], &pNode);
    }

    WriteTraceLog(LogType_StandardOut, LogLevel_Info, __FILE__, __LINE__, __FUNCTION__, "Can't find original recv message[CorrelationID:%s,MSG:%s]", szUUIDString, aMessageBody);
    return MQ_ERR_CANT_FIND_ORIG_MSG;
}



//dam test
int InitDirectLaneSendMQConfig()
{
    char szValueTemp[MAX_STRING_SIZE + 1];
/*
MQ_SERVER_IP=127.0.0.1
MQ_SERVER_PORT=5672
MQ_VIRTUALHOST=MessageService
#MQ_SERVICE_USER_ID=siadmin
MQ_SERVICE_USER_ID=daservice
MQ_SERVICE_PWD=neighbor21
MQ_MESSAGE_SEND_EXCHANGER_NAME=MessageSencExc
MQ_MESSAGE_RECV_QUEUE_NAME=MessageRecvQ
*/

    // Read MQ_CONF_SERVER_IP_KEY
    if (GetConfigValue(MQ_CONF_SERVER_IP_KEY, g_szMQServerIP, sizeof(g_szMQServerIP)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_IP_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVER_PORT_KEY
    if (GetConfigValue(MQ_CONF_SERVER_PORT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVER_PORT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nMQServerPort) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_SERVER_PORT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_VIRTUALHOST_KEY
    if (GetConfigValue(MQ_CONF_VIRTUALHOST_KEY, g_szMQVirtualHostname, sizeof(g_szMQVirtualHostname)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_VIRTUALHOST_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_USER_ID_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_USER_ID_KEY, g_szMQServiceID, sizeof(g_szMQServiceID)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_USER_ID_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_SERVICE_PWD_KEY
    if (GetConfigValue(MQ_CONF_SERVICE_PWD_KEY, g_szMQServicePWD, sizeof(g_szMQServicePWD)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_SERVICE_PWD_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY
    if (GetConfigValue(MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY, g_szMQMessageRecvQueuename, sizeof(g_szMQMessageRecvQueuename)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_MESSAGE_RECV_QUEUE_NAME_KEY);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_ORGANIZATION_CODE
    if (GetConfigValue(MQ_CONF_ORGANIZATION_CODE, g_szMQOrganizationCode, sizeof(g_szMQOrganizationCode)) != MQ_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_ORGANIZATION_CODE);
        return MQ_ERR_CONFIG;
    }

    // Read PROCESS_TYPE
    if (GetConfigValue(MQ_CONF_PROCESS_TYPE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration not found[Available Value:Gateway or Application]!");
        return MQ_ERR_CONFIG;
    }
    if (MappingStringToInt(szValueTemp, sizeof(szValueTemp), g_sProcTypeInfo, sizeof(g_sProcTypeInfo)/sizeof(struct VALUE_STRING_PAIR_INFO), &g_nProcessType) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Socket PROCESS_TYPE configuration error[ConfigFile:%s,Available Value:Server or Client]!", szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_SIZE_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_SIZE_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchSize) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_SIZE_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    // Read MQ_CONF_PREFETCH_COUNT_KEY
    if (GetConfigValue(MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp, sizeof(szValueTemp)) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue configuration not found[Configuration Key:%s]!", MQ_CONF_PREFETCH_COUNT_KEY);
        return MQ_ERR_CONFIG;
    }
    if (SafeStringToInt(szValueTemp, sizeof(szValueTemp), &g_nPrefetchCount) != CONF_ERR_SUCCESS)
    {
        WriteLog(LogType_ErrorOut, LogLevel_Error, "Message queue %s configuration error[ConfigFile:%s]!", MQ_CONF_PREFETCH_COUNT_KEY, szValueTemp);
        return MQ_ERR_CONFIG;
    }

    return MQ_ERR_SUCCESS;
}
