/*********************************************
 *           Code generated by the           *
 * Altibase Precompiler2(APRE) Ver.1(7.1.0.1.2)*
 *********************************************/

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <ulpLibInterface.h>

/* The variable _esqlopts specifies command-line options
 * {-n, -unsafe_null, stmt caching, ...}
 */
static short _esqlopts[10] = {0,0,1,0,0,0,0,0,0,0};

/*
 *========================================================================
 *
 *   시스템명 : 요금징수 시스템
 *   파 일 명 : wrk_data_hnd_main.pc
 *   작 성 자 : 조보용
 *   작 성 일 : 2012.09.13
 *   수 정 일 :
 *   설    명 : 근무자료 처리 프로세스
 *
 *========================================================================
*/

#include "wrk_data_def.h"
#include "MessageProcessor.h"
#include "taglnk_frame.h"

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

void    SIG_term(sig)
int     sig;
{
    int    ii;

    for (ii = 0; ii < 3; ii++)
    {
        /*  ROLLBACK WORK  */
{
    struct ulpSqlstmt ulpSqlstmt;
    memset(&ulpSqlstmt, 0, sizeof(ulpSqlstmt));
    ulpSqlstmt.stmttype = 19;
    ulpSqlstmt.stmtname = NULL;
    ulpSqlstmt.ismt = 1;
    ulpSqlstmt.numofhostvar = 0;
    ulpSqlstmt.statusptr = NULL;
    ulpSqlstmt.errcodeptr = NULL;
    ulpSqlstmt.isatomic = 0;
    ulpSqlstmt.iters = 0;
    ulpSqlstmt.sqlinfo = 3;
    ulpSqlstmt.scrollcur = 0;
    ulpSqlstmt.cursorscrollable = 0;
    ulpSqlstmt.cursorsensitivity = 1;
    ulpSqlstmt.cursorwithhold = 0;
    ulpSqlstmt.esqlopts    = _esqlopts;
    ulpSqlstmt.sqlcaerr    = &sqlca;
    ulpSqlstmt.sqlcodeerr  = &SQLCODE;
    ulpSqlstmt.sqlstateerr = ulpGetSqlstate();
    ulpDoEmsql( NULL, &ulpSqlstmt, NULL );
}

        if (!SQLCODE) break;
    }
    printf_s( " TERMINATE SIGNAL[%d]...... END\n", sig);

    exit(-1);
}

struct    writeque {           /* PACKET 변환 TABLE    */
    int    cmd;                /* command              */
    int    wrcheck;            /* Check                */
} writeque[] = {
   /* cmd */                  /* Check */
 { AZN_RQ_SEQ_NO            , 0 },  /* 0x0A 전송연번요구 */
 { AZN_COMM_CONTINUE_RQ     , 0 },  /* 0x0E 교신재개요구 */
 { AZN_TM_RQ                , 0 },  /* 0x0D 시각요구 */
 { AZN_DAY_CHNG             , 1 },  /* 0x1A 일변경 데이터 */
 { AZN_DAY_END              , 1 },  /* 0x1B 일마감 데이터 */
 { AZN_TBL_ORDER            , 0 },  /* 0x3A 테이블 요구 */
 { AZN_LANE_STAT            , 0 },  /* 0x8A 차로상태(STATUS) */
 { AZN_RPR_DATA             , 1 },  /* 0x7B 보수데이터 */
 { AZN_CLOSE_LANE_PASS      , 1 },  /* 0x8C 폐쇄차로 통행 데이터 */
 { AZN_SENS_DATA            , 0 },  /* 0x57 감지데이터(근무중차판) */
 { AZN_WRK_STRT             , 1 },  /* 0x58 근무개시보고 */
 { AZN_CARD_DATA            , 1 },  /* 0x90 출구처리 데이터 */
 { AZN_CAR_NO_DATA          , 1 },  /* 0x91 처리차량번호데이터 */
 { AZN_ENT_DATA             , 1 },  /* 0x27 입구처리 데이터 */
 { AZN_ISS_INFO             , 0 },  /* 0x21 발행정보 조회(부스에서 사용) */
 { AZN_PERD_TAG_SALE        , 0 },  /* 0x70 기간권 TAG 판매 등록 요청 */
 { AZN_PERD_TAG_SALE_RSPNS  , 0 },  /* 0x71 기간권 TAG 판매 등록 결과 */
 { AZN_RGSTR_TAG_CHRG       , 0 },  /* 0x73 기명권 충전 요청 */
 { AZN_RGSTR_TAG_CHRG_RSPNS , 0 },  /* 0x74 기명권 충전 응답 */
 { AZN_PREV_TR_INFO         , 0 },  /* 0x75 직전거래 내역 요청 */
 { AZN_PREV_TR_INFO_RSPNS   , 0 },  /* 0x76 직전거래 내역 응답 */
 { AZN_TAG_CNCL_CHRG        , 0 },  /* 0x77 기명권 충전 직전 취소 요청 */
 { AZN_TAG_CNCL_CHRG_RSPNS  , 0 },  /* 0x78 기명권 충전 직전 취소 응답 */
 { AZN_OUT_WRK_END_DATA     , 1 },  /* 0x4C 근무종료데이터(출구) */
 { AZN_IN_WRK_END_DATA      , 1 },  /* 0x4D 근무종료데이터(입구) */
 { AZN_LANE_LINE_TEST       , 0 },  /* 0x2E 차로측기동회선테스트 테이터 */
 { AZN_PLZ_LINE_TEST_REPLY  , 0 },  /* 0xDE 영업소측 기동회선테스트 테이터 응답 */
 { AZN_REQ_TAG_INFO         , 0 },  /* 0x31 TAG 정보 조회(ETCS/MLFF 전용) */
 { AZN_IN_WAY_SPCL_PRCS     , 1 },  /* 0x72 입구특별처리 데이터 */
 { AZN_OFC_PRCS_DATA        , 1 },  /* 0x8F 사무실 처리 데이터 */
 { AZN_ETCS_VIO_PRCS_DATA   , 1 },  /* 0x9F ETCS 출구 위반 입구 수납 */
 { -1                       , 0 }
};

/*
 *========================================================================
 *
 *   함 수 명 : proc_data()
 *   작 성 자 : 조보용
 *   작 성 일 : 2012.09.13
 *   수 정 일 :
 *   설    명 : Winfo Parsing & Processing
 *
 *========================================================================
*/

int proc_data(char *buffer, char *rep)
{
    struct rcv_lane_data *rd;
    int irv = 0;
    int jbcd;
    char ht[2];

    rd = (struct rcv_lane_data *)buffer;
    rep = szReplyBuffer;
printf("수신자료 [%s]\n", rd);
    snprintf(ht, 2+1, "%c%c", rd->buf[8], rd->buf[9]);
    jbcd = lb_string2hexa((unsigned char *)ht, 2);

    if(c_DbConnect()<0)
    {
        printf_s( "worksvr db connect error\n");
        /*  ROLLBACK  */
{
    struct ulpSqlstmt ulpSqlstmt;
    memset(&ulpSqlstmt, 0, sizeof(ulpSqlstmt));
    ulpSqlstmt.stmttype = 19;
    ulpSqlstmt.stmtname = NULL;
    ulpSqlstmt.ismt = 1;
    ulpSqlstmt.numofhostvar = 0;
    ulpSqlstmt.statusptr = NULL;
    ulpSqlstmt.errcodeptr = NULL;
    ulpSqlstmt.isatomic = 0;
    ulpSqlstmt.iters = 0;
    ulpSqlstmt.sqlinfo = 3;
    ulpSqlstmt.scrollcur = 0;
    ulpSqlstmt.cursorscrollable = 0;
    ulpSqlstmt.cursorsensitivity = 1;
    ulpSqlstmt.cursorwithhold = 0;
    ulpSqlstmt.esqlopts    = _esqlopts;
    ulpSqlstmt.sqlcaerr    = &sqlca;
    ulpSqlstmt.sqlcodeerr  = &SQLCODE;
    ulpSqlstmt.sqlstateerr = ulpGetSqlstate();
    ulpDoEmsql( NULL, &ulpSqlstmt, NULL );
}

        return -1;
    }
printf("처리종별 [%c][%c][%x]\n", rd->buf[8], rd->buf[9], jbcd);
    switch (jbcd)
    {
        case AZN_RQ_SEQ_NO:
            printf_s( " < 전송연번요구 > \n" );
            irv = proc_rq_snd_no(buffer, rep);



            break;
        case AZN_LANE_STAT:  /* 차로상태 */
            printf_s( " < 차로상태정보 > \n" );
            irv = proc_lane_stat_info(buffer);
            break;
        case AZN_RPR_DATA: /* 보수데이터 */
            printf_s( " <WRKPQUEUE 보수데이터 > \n" );
            irv = proc_rpr_data(buffer);
            break;
        case AZN_CLOSE_LANE_PASS: /* 폐쇄차로 */
            printf_s( " <WRKPQUEUE 폐쇄차로 > \n" );
            irv = proc_cls_pass(buffer);
            break;
        case AZN_SENS_DATA: /* 감지데이터 */
            printf_s( " < 감지데이터 > \n" );
            irv = proc_sens_data(buffer);
            break;
        case AZN_WRK_STRT:
        	printf_s( " <WRKPQUEUE 근무개시 > \n" );
            irv = proc_wrk_start(buffer);



            break;
        case AZN_OUT_WRK_END_DATA:
            printf_s( " <WRKPQUEUE 근무종료-출구 > \n" );
            irv = proc_outwrk_end(buffer);



            break;
        case AZN_IN_WRK_END_DATA:
            printf_s( " <WRKPQUEUE 근무종료-입구 > \n" );
            irv = proc_inwrk_end(buffer);



            break;
        case AZN_CARD_DATA:  /* 출구처리 */
            printf_s( " <WRKPQUEUE 출구처리    > \n" );
            irv = proc_out_way(buffer);
            break;
        case AZN_CAR_NO_DATA:  /* 처리차량번호 */
            printf_s( " <WRKPQUEUE 처리차량번호    > \n" );
            irv = proc_car_no(buffer);
            break;
        case AZN_ENT_DATA:  /* 입구처리 */
            printf_s( " <WRKPQUEUE 입구처리    > \n" );
            irv = proc_in_way(buffer);
            break;
        case AZN_ISS_INFO:
        	printf_s( " < 발행정보조회 > \n" );
        	irv = proc_iss_info(buffer, rep);



        	break;
        case AZN_PERD_TAG_SALE:
            printf_s( " < 기간권TAG판매등록요청 > \n" );
            irv = proc_tag_sale(buffer, rep);
            break;
        case AZN_REQ_TAG_INFO:
        	printf_s( " < TAG정보조회 > \n" );
        	irv = proc_tag_info(buffer, rep);



        	break;
        case AZN_PREV_TR_INFO:
            printf_s( " < 직전거래내역조회 > \n" );
            irv = proc_prev_tr_info(buffer, rep);



            break;
        case AZN_IN_WAY_SPCL_PRCS:  /* 입구특별처리 */
            printf_s( " <WRKPQUEUE 입구특별처리    > \n" );
            irv = proc_in_spcl(buffer);
            break;
        case AZN_OFC_PRCS_DATA:  /* 사무실 */
            printf_s( " <WRKPQUEUE 사무실    > \n" );
            irv = proc_ofc_data(buffer);
            break;
        case AZN_ETCS_VIO_PRCS_DATA:  /* 출구위반입구수납 */
            printf_s( " <WRKPQUEUE 출구위반입구수납    > \n" );
            irv = proc_etcs_vio(buffer);
            break;
        case AZN_DAY_CHNG:  /* 일변경 */
            printf_s( " <WRKPQUEUE 일변경    > \n" );
            irv = inst_day_chg(buffer);
            break;
        case AZN_DAY_END:  /* 일마감 */
            printf_s( " <WRKPQUEUE 일마감    > \n" );
            irv = inst_day_end(buffer);
            break;
      default:
          break;
    }
    if (irv < 0)
    {
        /*  ROLLBACK WORK; */
{
    struct ulpSqlstmt ulpSqlstmt;
    memset(&ulpSqlstmt, 0, sizeof(ulpSqlstmt));
    ulpSqlstmt.stmttype = 19;
    ulpSqlstmt.stmtname = NULL;
    ulpSqlstmt.ismt = 1;
    ulpSqlstmt.numofhostvar = 0;
    ulpSqlstmt.statusptr = NULL;
    ulpSqlstmt.errcodeptr = NULL;
    ulpSqlstmt.isatomic = 0;
    ulpSqlstmt.iters = 0;
    ulpSqlstmt.sqlinfo = 2;
    ulpSqlstmt.scrollcur = 0;
    ulpSqlstmt.cursorscrollable = 0;
    ulpSqlstmt.cursorsensitivity = 1;
    ulpSqlstmt.cursorwithhold = 0;
    ulpSqlstmt.esqlopts    = _esqlopts;
    ulpSqlstmt.sqlcaerr    = &sqlca;
    ulpSqlstmt.sqlcodeerr  = &SQLCODE;
    ulpSqlstmt.sqlstateerr = ulpGetSqlstate();
    ulpDoEmsql( NULL, &ulpSqlstmt, NULL );
}

        printf_s("[%d]irv : %d, Process Fail\n", jbcd, irv);
    }
    else
    {
        printf_s("irv : %d, COMMIT\n", irv);
        /*  COMMIT WORK; */
{
    struct ulpSqlstmt ulpSqlstmt;
    memset(&ulpSqlstmt, 0, sizeof(ulpSqlstmt));
    ulpSqlstmt.stmttype = 19;
    ulpSqlstmt.stmtname = NULL;
    ulpSqlstmt.ismt = 1;
    ulpSqlstmt.numofhostvar = 0;
    ulpSqlstmt.statusptr = NULL;
    ulpSqlstmt.errcodeptr = NULL;
    ulpSqlstmt.isatomic = 0;
    ulpSqlstmt.iters = 0;
    ulpSqlstmt.sqlinfo = 0;
    ulpSqlstmt.scrollcur = 0;
    ulpSqlstmt.cursorscrollable = 0;
    ulpSqlstmt.cursorsensitivity = 1;
    ulpSqlstmt.cursorwithhold = 0;
    ulpSqlstmt.esqlopts    = _esqlopts;
    ulpSqlstmt.sqlcaerr    = &sqlca;
    ulpSqlstmt.sqlcodeerr  = &SQLCODE;
    ulpSqlstmt.sqlstateerr = ulpGetSqlstate();
    ulpDoEmsql( NULL, &ulpSqlstmt, NULL );
}

    }

    /*  ROLLBACK  */
{
    struct ulpSqlstmt ulpSqlstmt;
    memset(&ulpSqlstmt, 0, sizeof(ulpSqlstmt));
    ulpSqlstmt.stmttype = 19;
    ulpSqlstmt.stmtname = NULL;
    ulpSqlstmt.ismt = 1;
    ulpSqlstmt.numofhostvar = 0;
    ulpSqlstmt.statusptr = NULL;
    ulpSqlstmt.errcodeptr = NULL;
    ulpSqlstmt.isatomic = 0;
    ulpSqlstmt.iters = 0;
    ulpSqlstmt.sqlinfo = 3;
    ulpSqlstmt.scrollcur = 0;
    ulpSqlstmt.cursorscrollable = 0;
    ulpSqlstmt.cursorsensitivity = 1;
    ulpSqlstmt.cursorwithhold = 0;
    ulpSqlstmt.esqlopts    = _esqlopts;
    ulpSqlstmt.sqlcaerr    = &sqlca;
    ulpSqlstmt.sqlcodeerr  = &SQLCODE;
    ulpSqlstmt.sqlstateerr = ulpGetSqlstate();
    ulpDoEmsql( NULL, &ulpSqlstmt, NULL );
}

    return(irv);
}

/*
 *========================================================================
 *
 *   함 수 명 : main()
 *   작 성 자 : 조보용
 *   작 성 일 : 2018.12.10
 *   수 정 일 :
 *   설    명 : 근무 자료 처리 DAEMON PROCESS(kmpsvr) MAIN
 *
 *========================================================================
*/
int main(int argc, char *argv[])
{
    char szProcessname[PATH_MAX + 1];
    char *pFindPointer;
    //int nPathIndicatorPos;
    int nResult;

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

    printf_s("szProcess Name : %s\n", szProcessname);

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

    GetExecPath(g_szProcessPath, sizeof(g_szProcessPath) - 1, argv[0]);
    printf_s("g_szProcess Name : %s\n", g_szProcessname);
    if (g_szProcessPath == NULL)
    {
        printf_s("Can't get process path[Cann't load config file]!\nExit Program!");
        return -1;
    }

    nResult = InitMessageProcessConfig();

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

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

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

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

    return 0;
}

int InitMessageProcessConfig()
{
    char szConfigfilename[PATH_MAX + 1];
    char szConfigfilenameTemp[PATH_MAX + 1];

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

    realpath(szConfigfilenameTemp, szConfigfilename);

    SafeStrCat(szConfigfilename, sizeof(szConfigfilename), "/", 1);
    SafeStrCat(szConfigfilename, sizeof(szConfigfilename), g_szProcessname, strnlen_s(g_szProcessname, sizeof(g_szProcessname)));
    SafeStrCat(szConfigfilename, sizeof(szConfigfilename), CONF_FILE_EXT, strnlen_s(CONF_FILE_EXT, sizeof(CONF_FILE_EXT)));

    printf_s("Process Name : %s\n", g_szProcessname);
    printf_s("Process Path : %s\n", g_szProcessPath);
    printf_s("Config Filename : %s\n", szConfigfilename);

    ReadConfigFile(true, szConfigfilename);

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

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

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

    return MP_ERR_SUCCESS;
}

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

    pThreadInfo = (struct THREAD_INFO *)pData;
    pnThreadExitCode = NULL;
    nResult = SafeMemoryAlloc((void **)&pnThreadExitCode, sizeof(int));
    //pnThreadExitCode = SafeAlloc(pnThreadExitCode, sizeof(int));
    *pnThreadExitCode = 0;
    while (pThreadInfo->bThreadLoop)
    {
        if (pThreadInfo->sPublishMQ.bMQConnection == false || pThreadInfo->sPublishMQ.bMQChannelOpen == false
            || pThreadInfo->sConsumeMQ.bMQConnection == false || pThreadInfo->sConsumeMQ.bMQChannelOpen == false)
        {
            if (ConnectProcessorMQ(pThreadInfo, true) != MQ_ERR_SUCCESS)
            {
                WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]ConnectProcessMQ is failed!", pThreadInfo->nThreadIndex);
                continue;
            }
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]ConnectProcessMQ is succeeded!!", pThreadInfo->nThreadIndex);
        }
printf("Reply Que[%s]\n", szReplyQueuename);
        //amqp_maybe_release_buffers(pThreadInfo->sMQConnectionState);
        nResult = GetMessageFromQueue(pThreadInfo->sConsumeMQ.sMQConnectionState, szExchangername, sizeof(szExchangername), szMessageBuffer, sizeof(szMessageBuffer), szCorrelationID, sizeof(szCorrelationID), szReplyQueuename, sizeof(szReplyQueuename), &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen, &nDeliveryTag, NULL);

        nMessageLength = strnlen_s(szMessageBuffer, sizeof(szMessageBuffer));
printf_s("\nDB 처리 전[%d][%s][%s]\n", nResult, szMessageBuffer, szReplyQueuename);
        if (nResult == MQ_ERR_SUCCESS)
        {
            printf_s("Message Received : %s\n", szMessageBuffer);
            //WriteRawCommLog(LogType_CommOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'C', "%s", szMessageBuffer);
            WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'C', g_szMQMessageRecvQueuename, "", szCorrelationID, nMessageLength, "%s", szMessageBuffer);
            WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]MQ Data Consume[Data:%s]!", pThreadInfo->nThreadIndex, szMessageBuffer);
            memset(szReplyBuffer, 0x00, sizeof(szReplyBuffer));
            nResult = proc_data(szMessageBuffer+45, szReplyBuffer);
printf_s("\nDB 처리 결과[%d]\nMessage[%s]\nReply[%s]\n", nResult, szMessageBuffer, szReplyBuffer);
            sprintf(szMessageBuffer+45, "%s", szReplyBuffer);
            //nResult = PutAckMessageToQueue(pThreadInfo->sMQConnectionState, szCorrelationID, szReplyQueuename, szMessageBuffer, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)), &pThreadInfo->bMQConnection, &pThreadInfo->bChannelOpen);
            amqp_basic_ack(pThreadInfo->sConsumeMQ.sMQConnectionState, 1, nDeliveryTag, 0);
            nResult = PutReplyMessageToQueue(pThreadInfo->sPublishMQ.sMQConnectionState, szCorrelationID, szReplyQueuename, szMessageBuffer, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)), &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen);
printf_s("\n응답처리 결과[%d][%d]\nMessage[%s]\nReplyQue[%s]\n", nResult, strnlen_s(szMessageBuffer, sizeof(szMessageBuffer)), szMessageBuffer, szReplyQueuename);
            WriteQueueLog(LogType_MQOut, LogLevel_Info, pThreadInfo->nThreadIndex, 'P', "", szReplyQueuename, szCorrelationID, nMessageLength, "%s", szMessageBuffer);
            WriteLog(LogType_StandardOut, LogLevel_Info, "[Thread:%02d]Response Publish[Queue:%s,UUID:%s,Data:%s]!", pThreadInfo->nThreadIndex, szReplyQueuename, szCorrelationID, szMessageBuffer);
        }
        else
        {
            WriteLog(LogType_StandardOut, LogLevel_Debug, "[Thread:%02d]MQ Data Consume Error!", pThreadInfo->nThreadIndex);
        }
    }
    CloseMQ(pThreadInfo->nThreadIndex, pThreadInfo->sPublishMQ.sMQConnectionState, &pThreadInfo->sPublishMQ.bMQConnection, &pThreadInfo->sPublishMQ.bMQChannelOpen, &pThreadInfo->sPublishMQ.sMQMutex);
    CloseMQ(pThreadInfo->nThreadIndex, pThreadInfo->sConsumeMQ.sMQConnectionState, &pThreadInfo->sConsumeMQ.bMQConnection, &pThreadInfo->sConsumeMQ.bMQChannelOpen, &pThreadInfo->sConsumeMQ.sMQMutex);

    return pnThreadExitCode;
}