worker.cpp

funambol windows mobile plugin source code, the source code is taken from the funambol site

/*
 * Copyright (C) 2003-2007 Funambol, Inc
 *
 * This program is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 2 as
 * published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY, TITLE, NONINFRINGEMENT or FITNESS FOR A PARTICULAR
 * PURPOSE.  See the GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA
 * 02111-1307  USA
 */

#include <stdlib.h>
#include "worker.h"
#include "s4jproxy.h"
#include "logutils.h"
#include "base/util/utils.h"
#include "notify/util.h"

#include "notify/addresschange.h"
#include "processutils.h"
#include "CTPManager.h"
#include "CTPParam.h"


#define MAX_ADDRESSCHANGE_TIME      60         // 5min = 300 seconds: big timeout for the addresschange sync

// Worker thread
extern "C" DWORD WINAPI NotifWorker(LPVOID lpv);

static bool syncinprogress = false;

// Handle of the worker thread
static HANDLE handle = 0;
// Connection socket
static SOCKET sock = 0;
static S4JProxy* s4jp = 0;
// Message buffer
static char *message = 0;
static int msglen = 0;
// Thread start time
time_t startTime;

/**
* Concatenate a char buf with specified len to the message buffer
*/
static void addBuffer(const char *buf, int len);

/**
* Close the communication socket client / server. Remove gracefully
* all the buffers and classes involved in the listening process.
*/
static void closeWorker();

/**
* Return the message
*/
const char *getMsg() { return message; }

/**
* Return the message lenght
*/
int getLen() { return msglen; }


/**
* Get the current time as Unix time_t
*/
static time_t getTime()
{
    SYSTEMTIME st;
    FILETIME ft;

    GetSystemTime(&st);
    SystemTimeToFileTime(&st, &ft);
    // Convert FILETIME to time_t
    __int64 llTmp;
    memcpy (&llTmp, &ft, sizeof (__int64));
    llTmp = (llTmp - 116444736000000000) / 10000000;
    return (time_t) llTmp;
}


/**
* Create the thread on the socket client / server waiting for the server
* messages
*/
DWORD startWorker(SOCKET s)
{
    // Check worker
    if (handle) {
        DebugPrint(TEXT("Another worker handle: %x\n"), handle);
        DWORD exitcode = 0;
        if(GetExitCodeThread(handle, &exitcode)) {
            if (exitcode == STILL_ACTIVE) {
                DebugPrint(TEXT("Worker thread still active\n"));
                if (syncinprogress) {
                    DebugPrint(TEXT("The sync is in progress\n"));
                    DebugPrint(TEXT("S4ND: connection busy\n"));
                    return ERROR_BUSY;
                }
                else if( (startTime - getTime()) <  ASK_TIMEOUT ) {
                    DebugPrint(TEXT("Request too close, wait\n"));
                    DebugPrint(TEXT("S4ND: connection busy\n"));
                    return ERROR_BUSY;
                }
            }
        }
        DebugPrint(TEXT("The sync is NOT in progress, closing thread\n"));
        if(sock){
            closesocket(sock);
            sock= 0;
        }
        Sleep(100);
        if(handle){
            DebugPrint(TEXT("Closing handle\n"));
            CloseHandle(handle);
            handle=0;
        }
    }
    // Clean-up dirty messages
    if(message){
        delete [] message;
        message=0;
    }
    msglen=0;

    // Start worker
    sock = s;
    s4jp = S4JProxy::getInstance();

    startTime = getTime();   // record starting time
    handle = CreateThread(NULL, 0, NotifWorker, (LPVOID*) sock, 0, NULL);
    if (handle == 0) {
        DebugPrint(L"S4ND:ERROR:CreateThread fails."
                L" GetLastError() returns <0x%08x>\n",
                GetLastError());
        // Release resources
        closeWorker();
    }

    DebugPrint(L"S4ND: new worker thread handle=<0x%08x>\n", handle);
    return ERROR_SUCCESS;
}

/**
* Close the communication socket client / server
*/
DWORD stopWorker()
{
    if(sock){
        closesocket(sock);
        sock= 0;
    }
    Sleep(100);
    if ( !handle )
        return ERROR_SUCCESS;
    else
        return ERROR_BUSY;
}

static void closeWorker()
{
    if(sock){
        closesocket(sock);
        sock= 0;
    }
    if(handle){
        CloseHandle(handle);
        handle=0;
    }
    S4JProxy::dispose();
    if(message){
        delete [] message;
        message=0;
    }
    msglen=0;

    DebugPrint(L"Worker end.\n");
    DebugEnd();
}

static void addBuffer(const char *buf, int len)
{
    if(!buf || !len)
        return;

    if(!message){
        message = new char[len+1];
        memcpy(message, buf, len);
        msglen = len;
    }
    else{
        char *tmp = new char[msglen+len+1];
        memcpy(tmp, message, msglen);
        memcpy(tmp+msglen, buf, len);
        msglen += len;
        delete [] message;
        message = tmp;
    }
}

/**
* Process the incoming server message and return a response.
* Return NULL if the message is not complete.
*/
const char *processPacket(const char *pkt, int pktlen)
{
    S4JProxy::RetCode code = S4JProxy::Error;
    static time_t last = getTime();

    time_t now = getTime();

    if ( (now - last) > 5 ){
        // reset message buffer
        delete [] message;
        message = 0;
    }

    // Message arrived
    DebugPrint(L"S4ND: %d bytes received\n", pktlen);

    addBuffer(pkt, pktlen);

    DebugPrint(L"S4ND: total message len %d\n", msglen);
    HexDump(message, msglen);

    code=s4jp->parsePkg0(message, msglen);
    DebugPrint(L"Return from parseMessage: %d\n", code);//XXX

    if (code != S4JProxy::Continue)
        return s4jp->getResponse();
    else
        return NULL;
}

/**
* Connection thread client/server. It waits for the server request,
* parses the message and if possible starts the synchronization of the
* server specified data source.
*/
extern "C" DWORD WINAPI NotifWorker(LPVOID lpv)
{
    SOCKET sock = (SOCKET)lpv;
    char buf[1501];
    int buflen=1500, pktlen;
    const char *response=NULL;

    while (!response) {
        pktlen = recv(sock, buf, buflen, 0);
        DebugPrint(L"Pktlen: %d\n", pktlen);    //XXX
        switch(pktlen) {
            case SOCKET_ERROR: {
                DWORD err = WSAGetLastError();
                if(err == WSAEMSGSIZE ){
                    // TODO: this case should be handled
                    DebugPrint(L"Message larger than %d bytes!", buflen);
                    response = "500 - Message too long\n";
                }
                else {
                    DebugPrint(L"Recv error: %d", err);
                    response = "500 - Network error\n"; // Try to respond anyway?
                }
                break;
            }
            case 0:
                // End of connection
                DebugPrint(L"Connection closed by peer.\n");
                response = "";  // Conn closed, no response
                break;
            default:
                response = processPacket(buf, pktlen);
        }
    }

    // If message is not empty, send it
    if(*response){
        DebugPrint(L"S4ND: Sending message <%S> to client\r\n", response);
        send(sock, response, strlen(response),0);
        // Start Sync
        if ( !s4jp->hasErrors() ) {
            closesocket(sock); // XXX
            syncinprogress=true;
            s4jp->sync();
            DebugPrint(L"Sync End.\n");
            syncinprogress=false;
        }
    }
    // Delete thread
    closeWorker();
    return 0;
}




/**
 * This is the main STP thread. Starts the notifyAddressChange sync
 * and returns the AN_ResponseCode: 0 if address accepted (success), 
 * otherwise > 0.
 */
DWORD WINAPI stpWorker(LPVOID lpv) {

    LOG.debug("Starting stpWorker thread");
    int ret = 0;
    LogLevel l = getLOGLevel();
    LOG.setLevel(l);

    AN_ResponseCode responseCode = notifyAddressChange(TEXT(APPLICATION_URI));
    if(responseCode == AN_AddressAccepted ){
        setServerNotified(true);
    }
    else {
        setServerNotified(false);
    }

    LOG.debug("Exiting stpWorker thread: response = %d", responseCode);
    return responseCode;
}



// -------------------- CTP --------------------------

/**
 * This is the main CTP thread. Manages the CTP connection using CTPManager,
 * starts threads to receive messages and to send 'ready' msg.
 */
DWORD WINAPI ctpWorker(LPVOID lpv) {

    LOG.debug("Starting ctpWorker thread");