ctpmanager.cpp

来自「funambol window mobile客户端源代码」· C++ 代码 · 共 898 行 · 第 1/2 页

    // Fill parameters (read values from config)
    CTPParam devId;
    devId.setParamCode(P_DEVID);
    devId.setValue(config.getDeviceId().c_str(), config.getDeviceId().size());
    authMsg.addParam(&devId);

    CTPParam username;
    username.setParamCode(P_USERNAME);
    username.setValue(config.getUsername().c_str(), config.getUsername().size());
    authMsg.addParam(&username);

    CTPParam cred;
    cred.setParamCode(P_CRED);
    // Create credentials from config props
    string credentials = createMD5Credentials();
    cred.setValue(credentials.c_str(), credentials.size());
    authMsg.addParam(&cred);

    string& fromValue = config.getUrlFrom();
    if (fromValue.size() > 0) {
        // FROM is used only after a JUMP status
        CTPParam from;
        from.setParamCode(P_FROM);
        from.setValue(fromValue.c_str(), fromValue.size());
        authMsg.addParam(&from);
    }

    LOG.info ("AUTH: devId='%s', user='%s', cred='%s'", config.getDeviceId().c_str(), 
                                                        config.getUsername().c_str(),
                                                        credentials.c_str() );

    // Send message
    return sendMsg(&authMsg);
}


/**
 * Sends a READY message to the Server.
 * A CTPMessage is filled with parameters from CTPConfig.
 * 'sendMsg()' method is used to send the message with the ctpSocket.
 * @return 0 if no errors
 */
int CTPManager::sendReadyMsg() { 

    // Fill CTPMessage members
    CTPMessage readyMsg;
    readyMsg.setGenericCommand(CM_READY);
    readyMsg.setProtocolVersion(CTP_PROTOCOL_VERSION);

    // Send message
    return sendMsg(&readyMsg);
}


/**
 * Sends a BYE message to the Server.
 * A CTPMessage is filled with parameters from CTPConfig.
 * 'sendMsg()' method is used to send the message with the ctpSocket.
 * @return 0 if no errors
 */
int CTPManager::sendByeMsg(){

    // Fill CTPMessage members
    CTPMessage byeMsg;
    byeMsg.setGenericCommand(CM_BYE);
    byeMsg.setProtocolVersion(CTP_PROTOCOL_VERSION);

    // Send message
    return sendMsg(&byeMsg);
}


/**
 * Sends a generic CTPMessage to the Server.
 * The socket must be already opened calling 'openConnection()'.
 * The CTPMessage passed must be already filled with all 
 * desired members, so a byte-array is formatted and sent to
 * the Server.
 * The 'ctpState' is set to CTP_STATE_WAITING_RESPONSE after the message 
 * is sent, as we always wait for a Server response for each sent msg.
 *
 * @param message  the CTPMessage ready to be sent
 * @return         0 if no errors
 * @note           no timeout is set for the socket 'send' operation
 */
int CTPManager::sendMsg(CTPMessage* message) {

    if (!message) {
        return 1;
    }

    int ret = 0;
    char* msg = message->toByte();
    int msgLength = message->getPackageLength();
    if (!ctpSocket) {
        LOG.error("sendMsg error: socket not initialized.");
        return 2;
    }

    // Debug the message to send.
    LOG.debug("Sending %d bytes:", msgLength);
    hexDump(msg, msgLength);


    ret = send(ctpSocket, msg, msgLength, 0);
    if (ret == SOCKET_ERROR) {
        DWORD errorCode = WSAGetLastError();
        LOG.error("CTPManager::sendMsg - send() error %d: %s", errorCode, createErrorMsg(errorCode));
        return errorCode;
    }
    else {
        LOG.debug("sendMsg - %d bytes sent", ret);
        ctpState = CTP_STATE_WAITING_RESPONSE;          // We wait for a Server response every msg sent!
        totalBytesSent += ret;
        LOG.debug("Total bytes sent since beginning: %d", totalBytesSent);

        // Will restore connection if no response in 60sec
        stopThread(cmdTimeoutThread);
        cmdTimeoutThread = CreateThread(NULL, 0, cmdTimeoutWorker, NULL, 0, NULL);
        if (!cmdTimeoutThread) {
            LOG.error("Error creating cmdTimeout thread: code 0x%08x", GetLastError());
        }
    }
    return 0;
}




/**
 * Receive a CTP message through the socket connection.
 * The message is parsed, a CTPMessage is filled and returned (the
 * CTPMessage is internally owned by CTPManager).
 * The message could be split into more packages, so we read the first 2 bytes
 * and keep receiving until the message is complete.
 * The ctpState is set to CTP_STATE_READY after the msg is received successfully.
 * 
 * @return  the received CTPMessage (pointer to internally owned object)
 * @note  this method calls winsock 'recv' function which is blocking
 */
CTPMessage* CTPManager::receiveStatusMsg(){

    char buffer[MAX_MESSAGE_SIZE], msg[MAX_MESSAGE_SIZE];
    DWORD errorCode    = 0;
    int totalBytes     = 0;
    int expectedLength = 0;

    if (receivedMsg) {
        delete receivedMsg;
        receivedMsg = NULL;
    }

    //
    // Receive socket message: could be split into more pkg
    //
    while (1) {
        LOG.info("Waiting for Server message...");
        int pkgLen = recv(ctpSocket, buffer, sizeof(buffer), 0);

        if (pkgLen == SOCKET_ERROR) {
            errorCode = WSAGetLastError();
            if (errorCode == WSAETIMEDOUT) {
                // recv timed out -> retry
                LOG.debug("Timeout error on recv() -> retry...");
                totalBytes = 0;
                continue;
            }
            else {
                // Socket error -> exit
                LOG.error("SOCKET recv() error %d: %s", errorCode, createErrorMsg(errorCode));
                goto finally;
            }
        }

        else if (pkgLen == 0) {
            LOG.debug("Socket connection closed by Server, exiting");
            goto finally;
        }

        else {
            if (totalBytes == 0) {      // first time
                expectedLength = extractMsgLength(buffer, pkgLen);
                if (!expectedLength) { goto finally; }
                expectedLength += 2;    // the first 2 bytes are the msg length
            }
            LOG.debug("Package received: %d bytes read (total = %d, expected = %d)", pkgLen, totalBytes+pkgLen, expectedLength);
            // Check if msg too big
            if (totalBytes+pkgLen >= MAX_MESSAGE_SIZE) {
                LOG.error("Message larger than %d bytes!", MAX_MESSAGE_SIZE);
                goto finally;
            }

            // Append bytes to the 'msg' array
            memcpy(&msg[totalBytes], buffer, pkgLen);
            totalBytes += pkgLen;

            // Check if msg is complete
            if (totalBytes < expectedLength) {
                LOG.debug("Message incomplete -> back to receive");
                continue;
            }
            else {
                LOG.debug("Message complete");
                ctpState = CTP_STATE_READY;             // ctpState back to 'ready'
                totalBytesReceived += totalBytes;
                
                // Debug the message received.
                LOG.debug("Received %d bytes:", totalBytes);
                hexDump(msg, totalBytes);
                LOG.debug("Total bytes received since beginning: %d", totalBytesReceived);
                break;
            }
        }
    }

    // Parse the message, receivedMsg is internally owned
    receivedMsg = new CTPMessage(msg, totalBytes);
    LOG.debug("status = 0x%02x", receivedMsg->getGenericCommand());

finally:
    stopThread(cmdTimeoutThread);       // Msg received or error, anyway kill the cmdTimeoutThread.
    return receivedMsg;
}



/**
 * This method is called when the CTP process is connected and ready to receive
 * notifications from the Server. Two threads are started here:
 *
 * 1. heartbeatWorker: used to send a READY message every 'ctpReady' seconds,
 *                     as an heartbeat for the CTP connection.
 * 2. receiveWorker  : keep listening to Server messages on the same socket.
 *                     Starts the sync when a notification is received.
 * Handles of both threads are internally owned by CTPManager, so threads
 * can be eventually terminated in case of errors.
 *
 * After the 2 threads have been started, we wait on the receiveThread 
 * (timeout = ctpConnTimeout, default = INFINITE) to be able to terminate the
 * CTP connection and restore it from scratch every 'ctpConnTimeout' seconds.
 * 
 * @return 0 if no errors
 * @note this method is blocked on the receiveThread (which is blocked on socket recv())
 *       and will exit only in case of:
 *       - socket errors
 *       - Server sends an error state
 *       - CTP in 'leaving state' (Client is closing CTP)
 */
int CTPManager::receive() {

    // Safe checks...
    if (!ctpSocket) {
        LOG.error("CTPManager::receive() error: no socket connection available");
        return -3;
    }
    if (stopThread(receiveThread, 1)) {
        LOG.debug("receiveThread killed");
    }
    if (stopThread(heartbeatThread, 1)) {
        LOG.debug("heartbeatThread killed");
    }

    //
    // Start thread to send 'ready' messages
    //
    heartbeatThread = CreateThread(NULL, 0, heartbeatWorker, (LPVOID*)ctpSocket, 0, NULL);
    if (!heartbeatThread) {
        LOG.error("Error creating heartbeat thread: code 0x%08x", GetLastError());
        return -2;
    }


    //
    // Start thread to receive messages from Server
    //
    receiveThread = CreateThread(NULL, 0, receiveWorker, (LPVOID*)ctpSocket, 0, NULL);
    if (!receiveThread) {
        LOG.error("Error creating receive thread: code 0x%08x", GetLastError());
        return -1;
    }


    //
    // Wait for receiveThread: it ends only in case of errors.
    // Use 'ctpConnTimeout' as timeout on this thread.
    //
    int ret = 0;
    DWORD timeout = getConfig()->getCtpConnTimeout();
    timeout *= 1000;
    if (timeout == 0) {
        timeout = INFINITE;
    }

    LOG.debug("Waiting for the receive thread to finish (timeout = %d sec)...", getConfig()->getCtpConnTimeout());
    DWORD waitResult = WaitForSingleObject(receiveThread, timeout);
    switch (waitResult) {

        // Thread exited: socket or Server error occurred -> out
        case WAIT_ABANDONED:
            LOG.debug("receiveThread abandoned");
        case WAIT_OBJECT_0: {
            DWORD exitcode = 0;
            GetExitCodeThread(receiveThread, &exitcode);
            LOG.debug("receiveThread ended with code %d", exitcode);
            ret = 0;
            break;
        }

        // Timeout: kill thread -> out.
        case WAIT_TIMEOUT: {
            LOG.debug("Timeout - receiveThread will now be terminated");
            TerminateThread(receiveThread, 0);
            ret = 1;
            break;
        }

        // Some error occurred (case WAIT_FAILED)
        default: {
            LOG.debug("Wait error on receiveThread");
            TerminateThread(receiveThread, 1);
            ret = 2;
            break;
        }
    }
    CloseHandle(receiveThread);
    receiveThread = NULL;

    // Also terminate the heartbeatThread
    if (stopThread(heartbeatThread)) {
        LOG.debug("heartbeatThread killed");
    }

    return ret;
}



/**
 * Utility to terminate a desired thread, setting its HANDLE to NULL.
 * @param thread   the HANDLE of the thread to be stopped
 * @param exitcode [optional] the desired exitcode
 * @return         true if the thread has been effectively terminated
 */
bool CTPManager::stopThread(HANDLE thread, DWORD exitcode) {

    if (thread) {
        TerminateThread(thread, exitcode);
        CloseHandle(thread);
        thread = NULL;
        return true;
    }
    return false;
}







/**
 * Formats a string for the 'cred' CTP param.
 * User credentials are encoded using MD5schema: 
 *   B64(MD5( B64(MD5("username":"password")):"clientNonce" ))
 * User parameters are retrieved from CTPConfig.
 * @return the credential string in b64 format.
 */
string CTPManager::createMD5Credentials() {

    string ret;
    char* credential = NULL;

    const char* username = config.getAccessConfig().getUsername();
    const char* password = config.getAccessConfig().getPassword();
    string clientNonce   = config.getCtpNonce();

    credential = MD5CredentialData(username, password, clientNonce.c_str());
    if (credential) {
        ret = credential;
        delete [] credential;
    }

    return ret;
}



/**
 * Utility function to retrieve the correspondant message for the WinSocket error code passed.
 * Pointer returned is allocated new, must be freed by caller.
 * @param errorCode  [OPTIONAL] the code of the last error
 * @return           the error message for the passed code (new allocated buffer)
 */
char* CTPManager::createErrorMsg(DWORD errorCode) {

    char* ret = NULL;
    if (!errorCode) {
        errorCode = GetLastError();
    }

    //
    // **** TODO: investigate why "FormatMessage()" does not work with winsocket ****
    // Error message is formatted manually for most common errors.
    //
    switch (errorCode) {
        case WSAETIMEDOUT:
            ret = stringdup("A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.");
            break;
        case WSAECONNREFUSED:
            ret = stringdup("No connection could be made because the target machine actively refused it.");
            break;
        case WSAECONNRESET:
            ret = stringdup("An existing connection was forcibly closed by the remote host.");
            break;
        case WSAESHUTDOWN:
            ret = stringdup("A request to send or receive data was disallowed because the socket had already been shut down in that direction with a previous shutdown call.");
            break;
        case WSAEHOSTDOWN:
            ret = stringdup("A socket operation failed because the destination host was down.");
            break;
        case WSAEHOSTUNREACH:
            ret = stringdup("A socket operation was attempted to an unreachable host.");
            break;
        default:
            ret = stringdup("Unknown socket error.");
            break;
    }

    //WCHAR* errorMessage = new WCHAR[512];
    //memset(errorMessage, 0, 512);
    //FormatMessage(
    //            FORMAT_MESSAGE_FROM_SYSTEM,
    //            NULL,
    //            errorCode,
    //            MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
    //            errorMessage,
    //            512,
    //            NULL);

    //if (!errorMessage || wcslen(errorMessage) == 0) {
    //    wsprintf(errorMessage, L"Unknown error.");
    //}
    //char* ret = toMultibyte(errorMessage);
    //if (errorMessage) delete [] errorMessage;

    return ret;
}