source: chevmsgr/trunk/msgclnt.cpp@ 14

#include "msgclnt.h"
#include "msg.hpp"

#include <algorithm>

#include <stdlib.h>

#include "cf/codec.h"

// --------------------------------------------------------------

typedef struct SMessageQWorkerArg
{
    cf::network::tcp * socket;
    MessageQ * messageQ;
} SMessageQWorkerArg;

typedef struct SCallbackWorkerArg
{
    MessageQ * messageQ;
    SCallback callback;
} SCallbackWorkerArg;

static int messageQworker(void * arg)
{
    SMessageQWorkerArg * inst = (SMessageQWorkerArg *)arg;

    Protocol::Message message;
    cf::size_t size;

    while (true)
    {
        try
        {
            size = 0;

            cf::bin raw = inst->socket->receive();
            size = raw.size();

            message.parse(raw.toString());
        }
        catch (cf::exception & e)
        {
            LOG(e.stackTrace());

            // closed
            if (size == 0)
            {
                // dummy logout
                message.parse(Protocol::Request().logout());
                inst->messageQ->push(message);
                break;
            }
        }

        inst->messageQ->push(message);
    }

    free(inst);

    return 0;
}

static inline SConversation toConversation(const Protocol::Message & message)
{
    SConversation c;
    std::string chat = message.get<std::string>(ProtocolType::MESSAGE);

    if (chat.length() > 0)
        c.message = cf::codec::hex::getInstance()->decode(chat).toString();

    c.sessid = message.get<std::string>(ProtocolType::SESSION_ID);
    c.sensitive = message.get<int>(ProtocolType::SENSITIVE);

    c.from = message.get<std::string>(ProtocolType::FROM);

    return c;
}

static inline SOpenSession toOpenSession(const Protocol::Message & message)
{
    SOpenSession o;

    o.sessid = message.get<std::string>(ProtocolType::SESSION_ID);
    o.idList = message.getList<std::string>(ProtocolType::ID_LIST);

    return o;
}

static int callbackWorker(void * arg)
{
    SCallbackWorkerArg * inst = (SCallbackWorkerArg *)arg;

    Protocol::Message message;

    while (true)
    {
        message = inst->messageQ->pop(ProtocolType::LISTEN, false);
        if (message.type() != ProtocolType::NONE)
            inst->callback.onListen(toConversation(message));

        message = inst->messageQ->pop(ProtocolType::OPEN_SESSION, false);
        if (message.type() != ProtocolType::NONE)
            inst->callback.onOpenSession(toOpenSession(message));

        message = inst->messageQ->pop(ProtocolType::LOGOUT, false);
        if (message.type() != ProtocolType::NONE)
            break;
    }

    free(inst);

    return 0;
}

// --------------------------------------------------------------

void MessageQ::push(const Protocol::Message & message)
{
    mutex.lock();
    messageQ.push_back(message);
    mutex.unlock();
}

Protocol::Message MessageQ::pop(const std::string & requestType, bool isWait)
{
    do
    {
        mutex.lock();
        if (messageQ.size() == 0)
        {
            mutex.unlock();
            continue;
        }

        Protocol::Message message = messageQ.front();

        if (message.type() == requestType)
        {
            messageQ.erase(messageQ.begin());
            mutex.unlock();
            return message;
        }
        mutex.unlock();
    } while (isWait);

    return Protocol::Message(); // return dummy NONE
}

// --------------------------------------------------------------

chev::chev()
    : listener(messageQworker),
      caller(callbackWorker)
{
}

chev::~chev()
{
    socket.close();
    listener.join();
    caller.join();
}

const std::string & chev::getLastError() const
{
    return error;
}

bool chev::connect(const std::string & host, const unsigned short port)
{
    try
    {
        socket.connect(host, port);

        SMessageQWorkerArg * arg = (SMessageQWorkerArg *)malloc(sizeof(SMessageQWorkerArg));
        if (!arg)
            return false;

        arg->socket = &socket;
        arg->messageQ = &messageQ;

        listener.start(arg);

        return true;
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return false;
    }
}

bool chev::join(const std::string & id, const std::string & pw, const std::string & sms)
{
    try
    {
        socket.send(request.join(id, pw, sms));

        return messageQ.pop(ProtocolType::JOIN).get<bool>(ProtocolType::RESULT);
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return false;
    }
}

bool chev::login(const std::string & id, const std::string & pw)
{
    try
    {
        request.setUserID(id);

        socket.send(request.login(pw));

        return messageQ.pop(ProtocolType::LOGIN).get<bool>(ProtocolType::RESULT);
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return false;
    }
}

bool chev::sms(const std::string & phone)
{
    try
    {
        socket.send(request.sms(phone));

        return messageQ.pop(ProtocolType::SMS).get<bool>(ProtocolType::RESULT);
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return false;
    }
}

bool chev::addFriend(const std::string & id)
{
    try
    {
        socket.send(request.addFriend(id));

        return messageQ.pop(ProtocolType::ADD_FRIEND).get<bool>(ProtocolType::RESULT);
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return false;
    }
}

std::vector<SFriend> chev::getFriendList()
{
    std::vector<SFriend> friendList;

    try
    {
        socket.send(request.getFriendList());

        friendList = messageQ.pop(ProtocolType::FRIEND_LIST).getFriendList();
    }
    catch (cf::exception & e)
    {
        LOG(e.what());
    }

    return friendList;
}

std::string chev::getSessionID(const std::vector<std::string> & idList)
{
    try
    {
        std::vector<std::string> toList = idList;
        toList.insert(toList.begin(), request.getUserID());

        std::string concat = joinStrings(toList);

        if (sessionMap.find(concat) == sessionMap.end())
        {
            socket.send(request.openSession(toList));

            Protocol::Message message = messageQ.pop(ProtocolType::OPEN_SESSION);

            sessionMap[concat] = message.get<std::string>(ProtocolType::SESSION_ID);
        }

        return sessionMap[concat];
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return "";
    }
}

bool chev::tell(const SConversation & conversation)
{
    try
    {
        SConversation c = conversation;

        if (c.message.length() > 0)
            c.message = cf::codec::hex::getInstance()->encode(c.message);

        socket.send(request.tell(c.sessid, c.message, c.sensitive));

        return messageQ.pop(ProtocolType::TELL).get<bool>(ProtocolType::RESULT);
    }
    catch (cf::exception & e)
    {
        LOG(e.what());

        return false;
    }
}

bool chev::listen(const SCallback & callback)
{
    SCallbackWorkerArg * arg = (SCallbackWorkerArg *)malloc(sizeof(SCallbackWorkerArg));
    if (!arg)
        return false;

    arg->messageQ = &messageQ;
    arg->callback = callback;

    caller.start(arg);

    return true;
}