putnLastBytes (&frame, (uint8 *)&netord, 4);

    netord = frameLength(frame);
    netord = htonl(netord);
    setnBytes(frame, 0, (uint8 *)&netord, 4);

#ifdef _DEBUG
info("create_cmpp_connect: connect pdu %d bytes:\n", frameLength(frame));
printFrame(INFO, frame, 0);
#endif

    deleteFrame(&cmcon->conpdu);

    cmcon->conpdu = frame;
    cmcon->concreated = 1;
    return 0;
}


int send_cmpp_connect (VTASK * vtask)
{
    CmppCon   * cmcon = NULL;
    int  times = 0;

    if (!vtask) return -1;

    cmcon = (CmppCon *)vtask->var;
    while ((!cmcon->concreated) && times++ < 5)
        create_cmpp_connect (vtask);

    if (!cmcon->concreated) return -1;

    if (sendTcpCon (cmcon->fd, cmcon->conpdu) < 0) {
        /* connection crashed */
        return -1;
    }

    startTimer (vtask, t_retry, TRETRY);
    ++cmcon->retrynum;

    return 0;
}



void cmpp_data_recv (VTASK * vtask, int fd, unsigned event)
{
    VTASK     * entity = NULL;
    CmppCon   * cmcon = NULL;
    CmppEntity * cment = NULL;
    FRAME_PTR   data = NULL;
    uint32      byteLen = 0;
    CmppPdu   * pdu = NULL;
    uint8       authismg [16];


    if (!vtask || fd == -1 || event != DF_READ)
        return;

    cmcon = (CmppCon *)vtask->var;

    if (cmcon->fd != fd) {
        error("cmpp_data_recv: fd mismatch.\n");
        return;
    }

    entity = (VTASK *)getEntity(vtask);
    cment = (CmppEntity *)entity->var;

    emptyFrame(cmcon->rcvbuf);

    if (recvTcpCon (fd, 4, &cmcon->rcvbuf) <= 0) {
        error("cmpp_data_recv: CMPP connection %s to ISMG %s:%d "
              "crashed while reading length.\n",
              vtask->name, cment->remote_host, cment->remote_port);
        vtask->state = cmpp_null;
        checkIdleConnection(vtask);

        startTimer(entity, t_build, TBUILD);
        return;
    }

    peeknBytes(cmcon->rcvbuf, 0, (uint8 *)&byteLen, 4);
    byteLen = ntohl(byteLen);

#ifdef _DEBUG
info("****data_recv: receive content %d bytes.\n", byteLen);
#endif

    byteLen -= 4;

    emptyFrame(cmcon->rcvbuf);

    if (recvTcpCon (fd, byteLen, &cmcon->rcvbuf) <= 0) {
        error("cmpp_data_recv: CMPP connection %s to ISMG %s:%d "
              "crashed while reading content.\n",
              vtask->name, cment->remote_host, cment->remote_port);
        vtask->state = cmpp_null;
        checkIdleConnection(vtask);

        startTimer(entity, t_build, TBUILD);
        return;
    }

#ifdef _DEBUG
info("***data_recv: recv %d bytes content:\n", frameLength(cmcon->rcvbuf));
printFrame(INFO, cmcon->rcvbuf, 0);
#endif

    /* prevent the getnLastBytes from deleting the frame whiel decoding */
    if (byteLen == frameLength(cmcon->rcvbuf))
        putLastByte(&cmcon->rcvbuf, '\0');

    pdu = cmpdu_decode (&cmcon->rcvbuf);
    if (!pdu) {
        error ("cmpp_data_recv: received data format error.\n");
        return;
    }

    switch (pdu->command_id) {
    case CMPP_CONNECT_RESP:
#ifdef _DEBUG
info("cmpp sp: got CMPP_CONNECT_RESP.\n");
#endif
        if (vtask->state != cmpp_handshaking) {
            warning("Warning: received CMPP_CONNECT_RESP not in "
                    "cmpp_handshaking state %d.\n", vtask->state);
            break;
        }

        switch (pdu->ConnectResp.status) {
        case 0:
            /* first authenticate ISMG via MD5 value contrast */
            putLastByte (&data, pdu->ConnectResp.status);
            putnLastBytes (&data, cmcon->authsrc, 16);
            putnLastBytes (&data, cmcon->sharedsecret, 
                                   strlen(cmcon->sharedsecret));
            Md5Val (bytePointer(data), frameLength(data), authismg);

#ifdef _DEBUG
info("***recv_data, to be MD5 is %d bytes.\n", frameLength(data));
printFrame(INFO, data, 0);
info("***recv_data, ISMG authen MD5 value:\n");
printOctet(INFO, authismg, 0, 16, 0);
#endif

            deleteFrame(&data);

            if (0) {//memcmp (authismg, pdu->ConnectResp.auth_ismg, 16) != 0) {
                tolog ("Authentication to ISMG [%s:%d] failed while "
                       "handling CONNECT_RESP. "
                       "\n\tSPID:[%s] SharedSecret:[%s] "
                       "ProtocolVersion:[%d.%d] TimeStamp:[%010d]\n",
                       cment->remote_host, cment->remote_port,
                       cmcon->sp_id, cmcon->sharedsecret,
                       (cmcon->version >> 4) & 0x0f, (cmcon->version & 0x0f),
                       cmcon->timestamp);
                
                vtask->state = cmpp_handshaking;
            } else {
                tolog ("CMPP Connect to ISMG [%s:%d] succeeds. "
                       "\n\tSPID:[%s] SharedSecret:[%s] "
                       "ProtocolVersion:[%d.%d] TimeStamp:[%010d]\n",
                       cment->remote_host, cment->remote_port,
                       cmcon->sp_id, cmcon->sharedsecret,
                       (cmcon->version >> 4) & 0x0f, (cmcon->version & 0x0f),
                       cmcon->timestamp);
                
                cmcon->retrynum = 0;
                vtask->state = cmpp_ready;

                /*cment->total_avail += cmcon->sliding_len;
                SetEvent(cment->hAvailEvent, 1);*/

                if (cmcon->is_mo) {
                    sk_push(cment->mo_list, vtask);
                } else {
                    cment->total_avail += cmcon->sliding_len;
                    SetEvent(cment->hAvailEvent, 1);
                    sk_push(cment->init_list, vtask);
                }

                if (sk_num(cment->init_list) < cment->mt_num ||
                    sk_num(cment->mo_list) < cment->mo_num) {
#ifdef _DEBUG
info("cmpp_data_recv: %d MT %d MO connections < %d total num,"
         " now start timer t_build.\n",
          sk_num(cment->init_list), sk_num(cment->mo_list), cment->cmpp_con_num);
#endif
                    startTimer(entity, t_build, TBUILD);
                }

                /*for (i=0; i<cmcon->sliding_len; i++)
                    sem_post (&cment->sem_submit);*/

                /* start connection test message timer */
                if (!cmcon->is_mo)
                    startTimer (vtask, t_acttest, TACTTEST);

                /* start the distribute procedure hook */
                startHook(cment->hDistribute);

                /* stop the retry timer that retransmit CONNECT message */
                stopTimer(vtask, t_retry);
            }
            break;
        case 1:
            tolog ("Connect_Resp indicates: connect message "
                   "structure error.\n");
            break;
        case 2:
            tolog ("Connect_Resp indicates: invalid source address [%s:%d]"
                   " of connect message.\n", inet_ntoa(cmcon->local_ip),
                   cmcon->local_port);
            break;
        case 3:
            tolog ("Connect_Resp indicates: authentication to SP "
                   "[ID:%s  SharedSecret:%s] is illegal.\n",
                   cmcon->sp_id, cmcon->sharedsecret);
            break;
        case 4:
            tolog ("Connect_Resp indicates: CMPP implementation "
                   "version [%d.%d] is too high.\n",
                   (cmcon->version >> 4) & 0x0f, (cmcon->version & 0x0f));
            break;
        default:
            tolog ("Connect_Resp indicates: unknown error.\n");
            break;
        }
  
        if (vtask->state != cmpp_ready) {
            if (cmcon->retrynum <= MAXRETRY && send_cmpp_connect(vtask) >= 0) {
                    vtask->state = cmpp_handshaking;
#ifdef _DEBUG
info("\ngot submit response not authenticated, send cmpp_connect %d times.\n", cmcon->retrynum);
#endif
            } else {
                if (cmcon->retrynum > MAXRETRY)
                    error("Got Submit Response: after %d times retrying to "
                          "sending CONNECT, system decides to terminate "
                          "this CMPP connection.\n", MAXRETRY);
                else
                    error("Got Submit Response: TCP connection crashed "
                          "while sending CMPP Connect.\n");

                vtask->state = cmpp_null;
                checkIdleConnection(vtask);

                startTimer(entity, t_build, TBUILD);
#ifdef _DEBUG
info("\ngot submit response not authenticated, send cmpp_connect error or "
           "retry exceed numer %d times.\n", cmcon->retrynum);
#endif
                break;
            }
        }
        break;

    case CMPP_TERMINATE:
        issue_terminate_resp (vtask, pdu);

        vtask->state = cmpp_null;
        checkIdleConnection(vtask);
        startTimer(entity, t_build, TBUILD);

        tolog("ISMG %s:%d terminate the CMPP connection %s.\n",
              cment->remote_host, cment->remote_port, vtask->name);
        break;

    case CMPP_TERMINATE_RESP:
        break;

    case CMPP_SUBMIT_RESP:
        if (handle_submit_resp (vtask, pdu) < 0) {
            error("cmpp_data_recv: submit response with sequence id %ld "
                  "has no corresponding submit.\n", pdu->sequence_id);
        } else {
            /* pdu has been stored, prevent it from being deleted. */
            pdu = NULL;
        }
        break;

    case CMPP_DELIVER:
        if (handle_deliver (vtask, pdu) < 0) {
            error("cmpp_data_recv: MO deliver message with sequence id %ld "
                  "has not been handled correctly.\n", pdu->sequence_id);

        } else {
            /* pdu has been stored, prevent it from being deleted. */
            pdu = NULL;
        }
        break;

    case CMPP_QUERY_RESP:
        break;

    case CMPP_CANCEL_RESP:
        break;

    case CMPP_ACTIVE_TEST:
        if (issue_active_test_resp (vtask, pdu) < -1) {
            error("cmpp_data_recv: CMPP active test message with sequence "
                  "id %ld "
                  "has not been handled correctly.\n", pdu->sequence_id);

        }
        break;

    case CMPP_ACTIVE_TEST_RESP:
        if (handle_active_test_resp(vtask, pdu) < 0) {
            error("cmpp_data_recv: CMPP ActiveTestResponse message with "
                  "sequence id %ld "
                  "has not been handled correctly.\n", pdu->sequence_id);

        } 
        break;
    }


    cmpdu_free (pdu);

    checkIdleConnection (vtask);
    return;
}