antEnvBaseMsg.cc

Go to the documentation of this file.

/*
 * This file is part of openSDK.
 *
 * Copyright (C) 2006-2007 openSDK team
 *
 * openSDK is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; version 2.
 *
 * openSDK is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY 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 openSDK; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 *
 *     $Id: antEnvBaseMsg.cc,v 1.24 2007/07/16 16:17:40 nuno-lopes Exp $
 */

#include "OpenSdkThread.h"

#include <OPENR/OSyslog.h>
#include <ant.h>
#include <EndpointTypes.h>
#include <TCPEndpointMsg.h>
#include <UDPEndpointMsg.h>
#include <DNSEndpointMsg.h>


// undefining Open-R stuff
#undef IPPROTO_IP    /* dummy for IP */
#undef IPPROTO_ICMP  /* control message protocol */
#undef IPPROTO_IGMP  /* group mgmt protocol */
#undef IPPROTO_GGP   /* gateway^2 (deprecated) */
#undef IPPROTO_TCP   /* tcp */
#undef IPPROTO_EGP   /* exterior gateway protocol */
#undef IPPROTO_PUP   /* pup */
#undef IPPROTO_UDP   /* user datagram protocol */
#undef IPPROTO_IDP   /* xns idp */
#undef IPPROTO_TP    /* tp-4 w/ class negotiation */
#undef IPPROTO_EON   /* ISO cnlp */
#undef IPPROTO_ENCAP /* encapsulation header */
#undef IPPROTO_RAW   /* raw IP packet */
#undef IPPROTO_MAX   
#undef IPPORT_RESERVED
#undef IPPORT_USERRESERVED

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <unistd.h>
#include <errno.h>

#include <pthread.h>

#include <map>
#include "OpenSdkEndpoint.h"
#include "OpenSdkEndpointInfo.h"
#include <opensdkAPI.h>
#include "antSharedBufferManager.h"

using namespace std;

typedef std::map<int, OpenSdkEndpointInfo*> EndpointInfoMap;

static EndpointInfoMap endpointMap = EndpointInfoMap();
static int nextEndpointId = 0;
static pthread_mutex_t mapMut = PTHREAD_MUTEX_INITIALIZER; // used to sychronize access to the previous 2 variables, and its contents

// remember which ports we are already listening
struct portsStruct {
        int socket;
        pthread_mutex_t mutex;

        static inline portsStruct getNew(int s) {
                portsStruct x;
                x.socket = s;
                pthread_mutex_init(&x.mutex, NULL);
                return x;
        }
};
static map<short, portsStruct> PortsMap;
static pthread_mutex_t PortsMutex = PTHREAD_MUTEX_INITIALIZER;


// helper functions
static TCPEndpointError TCP_error(void);
static TCPEndpointError TCP_state_error(OpenSdkEndpointInfo::State state);
static UDPEndpointError UDP_error(void);
static UDPEndpointError UDP_state_error(OpenSdkEndpointInfo::State state);

static void* tcpListenThreadFunc(void *arg);
static void* tcpReceiveThreadFunc(void *arg);
static void* tcpSendThreadFunc(void *arg);
static void* tcpConnectThreadFunc(void *arg);
static void* tcpCloseThreadFunc(void *arg);

static void* udpReceiveThreadFunc(void *arg);
static void* udpSendThreadFunc(void *arg);
static void* udpCloseThreadFunc(void *arg);

static void* dnsGetHostByNameThreadFunc(void *arg);


antStackRef::antStackRef(char* name)
{
        if (strcmp(name, "IPStack")) {
                OSYSLOG1((osyslogERROR, "antStackRef::antStackRef(): protocol stack not supported: '%s'", name));
        }
}

antStackRef::~antStackRef()
{
        //TODO
}


antError antEnvMsg::Reply()
{
        //TODO
        return ANT_FAIL;
}


antError antEnvMsg::Send(antStackRef _target, OID& _sender, int32 _selector, int _msgSize)
{
        OpenSdkEndpoint *endpoint = (OpenSdkEndpoint*)(&(this->dstModule));

        pthread_mutex_lock(&mapMut);
        OpenSdkEndpointInfo *osei = endpointMap[(int)endpoint->endpointId];
        osei->entry.Set(_sender, _selector);
        pthread_mutex_unlock(&mapMut);

        switch (endpoint->protocol) {

                case EndpointType_TCP:

                        switch(this->dstOperation) {

                                case TCPEndpoint_Operation_Bind:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_Listen:
                                {
                                        TCPEndpointListenMsg *listenMsg = (TCPEndpointListenMsg*)this;
                                        struct sockaddr_in listeningAddress;
                                        int sret;

                                        if (osei->state != OpenSdkEndpointInfo::NEW && osei->state != OpenSdkEndpointInfo::CLOSED) {
                                                listenMsg->error = TCP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        // check if we are already listening to this port
                                        pthread_mutex_lock(&PortsMutex);
                                        if (PortsMap.find(listenMsg->lPort) != PortsMap.end()) {
                                                osei->server_socket = PortsMap[listenMsg->lPort].socket;
                                                goto tcp_listen_skip;
                                        }

                                        osei->server_socket = socket(PF_INET, SOCK_STREAM, 0);

                                        if (osei->server_socket < 0) {
                                                listenMsg->error = TCP_error();
                                                pthread_mutex_unlock(&PortsMutex);
                                                return ANT_FAIL;
                                        }

                                        memset(&listeningAddress, 0, sizeof(listeningAddress));
                                        listeningAddress.sin_family      = AF_INET;
                                        listeningAddress.sin_addr.s_addr = listenMsg->lAddress.Address();
                                        listeningAddress.sin_port        = htons(listenMsg->lPort);

                                        sret = bind(osei->server_socket, (const struct sockaddr *)&listeningAddress, sizeof(listeningAddress));
                                        if (sret != 0) {
                                                listenMsg->error = TCP_error();
                                                pthread_mutex_unlock(&PortsMutex);
                                                return ANT_FAIL;
                                        }

                                        sret = listen(osei->server_socket, 42);
                                        if (sret != 0) {
                                                listenMsg->error = TCP_error();
                                                pthread_mutex_unlock(&PortsMutex);
                                                return ANT_FAIL;
                                        }

                                        // save the socket fd for other threads
                                        PortsMap[listenMsg->lPort] = portsStruct::getNew(osei->server_socket);
tcp_listen_skip:
                                        osei->listen_mutex  = &PortsMap[listenMsg->lPort].mutex;
                                        pthread_mutex_unlock(&PortsMutex);

                                        osei->msg   = new TCPEndpointListenMsg(*listenMsg);
                                        osei->state = OpenSdkEndpointInfo::LISTENING;

                                        if (OpenSdkThread::runFunction(tcpListenThreadFunc, osei) == 0) {
                                                listenMsg->error = TCP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                listenMsg->error = TCP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;


                                case TCPEndpoint_Operation_Connect:
                                {
                                        TCPEndpointConnectMsg *connectMsg = (TCPEndpointConnectMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::NEW && osei->state != OpenSdkEndpointInfo::CLOSED) {
                                                connectMsg->error = TCP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei->socket = socket(PF_INET, SOCK_STREAM, 0);
                                        if (osei->socket < 0) {
                                                connectMsg->error = TCP_error();
                                                return ANT_FAIL;
                                        }

                                        osei->ip    = connectMsg->fAddress;
                                        osei->port  = connectMsg->fPort;
                                        osei->state = OpenSdkEndpointInfo::CONNECTING;
                                        osei->msg   = new TCPEndpointConnectMsg(*connectMsg);

                                        if (OpenSdkThread::runFunction(tcpConnectThreadFunc, osei) == 0) {
                                                connectMsg->error = TCP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                connectMsg->error = TCP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;


                                case TCPEndpoint_Operation_Send:
                                {
                                        TCPEndpointSendMsg *sendMsg = (TCPEndpointSendMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::ACTIVE) {
                                                sendMsg->error = TCP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei->buffer = sendMsg->buffer;
                                        osei->size   = sendMsg->size;
                                        osei->msg    = new TCPEndpointSendMsg(*sendMsg);

                                        if (OpenSdkThread::runFunction(tcpSendThreadFunc, osei) == 0) {
                                                sendMsg->error = TCP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                sendMsg->error = TCP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;


                                case TCPEndpoint_Operation_Receive:
                                {
                                        TCPEndpointReceiveMsg *receiveMsg = (TCPEndpointReceiveMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::ACTIVE) {
                                                receiveMsg->error = TCP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei->buffer = receiveMsg->buffer;
                                        osei->size   = receiveMsg->sizeMax;
                                        osei->msg    = new TCPEndpointReceiveMsg(*receiveMsg);

                                        if (OpenSdkThread::runFunction(tcpReceiveThreadFunc, osei) == 0) {
                                                receiveMsg->error = TCP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                receiveMsg->error = TCP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;


                                case TCPEndpoint_Operation_Close:
                                {
                                        TCPEndpointCloseMsg *closeMsg = (TCPEndpointCloseMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::ACTIVE) {
                                                closeMsg->error = TCP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei->msg = new TCPEndpointCloseMsg(*closeMsg);

                                        if (OpenSdkThread::runFunction(tcpCloseThreadFunc, osei) == 0) {
                                                closeMsg->error = TCP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                closeMsg->error = TCP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;


                                case TCPEndpoint_Operation_SetBlockingMode:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_ReadyForRead:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_ReadyForWrite:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_BacklogListen:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_Accept:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_BacklogQueues:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                case TCPEndpoint_Operation_Address:
                                {
                                        // TODO
                                        return ANT_FAIL;
                                }
                                break;


                                default:
                                        OSYSLOG1((osyslogERROR, "antEnvMsg::Send: Invalid TCP message type (%d)", this->dstOperation));
                                break;

                        }
                        break;

                case EndpointType_UDP:

                        switch(this->dstOperation) {

                                case UDPEndpoint_Operation_Send:
                                {
                                        UDPEndpointSendMsg *sendMsg = (UDPEndpointSendMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::ACTIVE) {
                                                sendMsg->error = UDP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei = new OpenSdkEndpointInfo(*osei);

                                        osei->buffer = sendMsg->buffer;
                                        osei->size   = sendMsg->size;
                                        osei->msg    = new UDPEndpointSendMsg(*sendMsg);

                                        if (OpenSdkThread::runFunction(udpSendThreadFunc, osei) == 0) {
                                                sendMsg->error = UDP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                sendMsg->error = UDP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;

                                case UDPEndpoint_Operation_Receive:
                                {
                                        UDPEndpointReceiveMsg *receiveMsg = (UDPEndpointReceiveMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::ACTIVE) {
                                                receiveMsg->error = UDP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei = new OpenSdkEndpointInfo(*osei);

                                        osei->buffer = receiveMsg->buffer;
                                        osei->size   = receiveMsg->size;
                                        osei->msg    = new UDPEndpointReceiveMsg(*receiveMsg);

                                        if (OpenSdkThread::runFunction(udpReceiveThreadFunc, osei) == 0) {
                                                receiveMsg->error = UDP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                receiveMsg->error = UDP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;

                                case UDPEndpoint_Operation_Close:
                                {
                                        UDPEndpointCloseMsg *closeMsg = (UDPEndpointCloseMsg*)this;

                                        if (osei->state != OpenSdkEndpointInfo::ACTIVE) {
                                                closeMsg->error = UDP_state_error(osei->state);
                                                return ANT_FAIL;
                                        }

                                        osei      = new OpenSdkEndpointInfo(*osei);
                                        osei->msg = new UDPEndpointCloseMsg(*closeMsg);

                                        if (OpenSdkThread::runFunction(udpCloseThreadFunc, osei) == 0) {
                                                closeMsg->error = UDP_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                closeMsg->error = UDP_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;

                                default:
                                        OSYSLOG1((osyslogERROR, "antEnvMsg::Send: Invalid UDP message type (%d)", this->dstOperation));
                                break;
                        }
                break;

                case EndpointType_DNS:

                        switch(this->dstOperation) {
                                case DNSEndpoint_Operation_GetHostByName:
                                {
                                        DNSEndpointGetHostByNameMsg *msg = (DNSEndpointGetHostByNameMsg*)this;

                                        osei->msg      = new DNSEndpointGetHostByNameMsg(*msg);
                                        osei->hostname = strdup(msg->name);

                                        if (OpenSdkThread::runFunction(dnsGetHostByNameThreadFunc, osei) == 0) {
                                                msg->error = DNS_SUCCESS;
                                                return ANT_SUCCESS;

                                        } else {
                                                msg->error = DNS_FAIL;
                                                return ANT_FAIL;
                                        }
                                }
                                break;

                                case DNSEndpoint_Operation_Close:
                                {
                                        // do nothing
                                }
                                break;

                                default:
                                        OSYSLOG1((osyslogERROR, "antEnvMsg::Send: Invalid DNS message type (%d)", this->dstOperation));
                        }
                break;

                default:
                        OSYSLOG1((osyslogERROR, "antEnvMsg::Send: Invalid protocol (%d)", endpoint->protocol));
                break;

        } // switch(endpoint->protocol)

        return ANT_FAIL;
}


antError antEnvMsg::Call(antStackRef _dest, int _msgSize)
{
        switch (this->dstSelector) {
        case Entry_antEnv_External:

        switch (this->dstOperation) {

                case antEnv_Operation_CreateEndpoint:
                {
                        antEnvCreateEndpointMsg *cEndPointMsg = (antEnvCreateEndpointMsg*)this;

                        pthread_mutex_lock(&mapMut);
                        int endpointId = nextEndpointId++;
                        OpenSdkEndpointInfo *osei = new OpenSdkEndpointInfo();
                        endpointMap[endpointId] = osei;
                        pthread_mutex_unlock(&mapMut);

                        antModuleRef endpoint((void*)endpointId, cEndPointMsg->protocol);

                        cEndPointMsg->moduleRef = endpoint;
                        cEndPointMsg->error     = ANT_SUCCESS;
                        return ANT_SUCCESS;

                }
                break;


                case antEnv_Operation_CreateSharedBuffer:
                {
                        antEnvCreateSharedBufferMsg *cSharedBufferMsg = (antEnvCreateSharedBufferMsg*)this;

                        void *bufferPtr = malloc(cSharedBufferMsg->size);
                        if (bufferPtr == NULL) {
                                cSharedBufferMsg->error = ANT_NOMEMORY;
                                return ANT_NOMEMORY;

                        } else {
                                antSharedBuffer asb      = antSharedBufferManager::getSharedBuffer(bufferPtr, cSharedBufferMsg->size);
                                cSharedBufferMsg->buffer = asb;
                                cSharedBufferMsg->error  = ANT_SUCCESS;
                                return ANT_SUCCESS;
                        }

                }
                break;

                case antEnv_Operation_DestroySharedBuffer:
                {
                        antEnvDestroySharedBufferMsg *msg = (antEnvDestroySharedBufferMsg*)this;
                        free(msg->buffer.GetAddress());
                        msg->error = ANT_SUCCESS;
                        return ANT_SUCCESS;

                }
                break;

                default:
                {

                        // operation depends on endpoint and on the protocol type
                        OpenSdkEndpoint *endpoint = (OpenSdkEndpoint*)(&(this->dstModule));

                        switch (endpoint->protocol) {

                                case EndpointType_UDP:
                                {

                                        switch(this->dstOperation) {

                                                case UDPEndpoint_Operation_Bind:
                                                {
                                                        UDPEndpointBindMsg * udpBindMsg = (UDPEndpointBindMsg*)this;

                                                        struct sockaddr_in udpAddress;
                                                        memset(&udpAddress, 0, sizeof(udpAddress));
                                                        udpAddress.sin_family      = AF_INET;
                                                        udpAddress.sin_addr.s_addr = udpBindMsg->address.Address();
                                                        udpAddress.sin_port        = htons(udpBindMsg->port);

                                                        pthread_mutex_lock(&mapMut);
                                                        OpenSdkEndpointInfo *osei = endpointMap[(int)endpoint->endpointId];
                                                        pthread_mutex_unlock(&mapMut);

                                                        osei->socket = socket(PF_INET, SOCK_DGRAM, 0);

                                                        if (osei->socket < 0) {
                                                                udpBindMsg->error = UDP_error();
                                                                return ANT_FAIL;
                                                        }

                                                        int bindRet = bind(osei->socket, (const struct sockaddr *)&udpAddress, sizeof(udpAddress));

                                                        if (bindRet != 0) {
                                                                udpBindMsg->error = UDP_error();
                                                                return ANT_FAIL;
                                                        }

                                                        osei->state = OpenSdkEndpointInfo::ACTIVE;
                                                        udpBindMsg->error = UDP_SUCCESS;

                                                        return ANT_SUCCESS;

                                                }
                                                break;

                                                default:
                                                        OSYSLOG1((osyslogERROR, "antEnvMsg::Call: invalid UDP dstOperation (%d)", this->dstOperation));
                                                        return ANT_FAIL;

                                        } // switch dstOperation

                                }
                                break;

                                default:
                                        OSYSLOG1((osyslogERROR, "antEnvMsg::Call : invalid protocol (%d)", endpoint->protocol));
                                        return ANT_FAIL;
                        }
                }
        }
        break;

        case Entry_antEnv_Initialize:
        {
                switch (this->dstOperation) {
                        case antEnv_Operation_InitSetParam:
                        {
                                return ANT_FAIL;
                        }
                        break;

                        case antEnv_Operation_InitStart:
                        {
                                return ANT_FAIL;
                        }
                        break;

                        case antEnv_Operation_InitGetParam:
                        {
                                return ANT_FAIL;
                        }
                        break;

                        default:
                                OSYSLOG1((osyslogERROR, "antEnvMsg::Call: invalid init dstOperation (%d)", this->dstOperation));
                                return ANT_FAIL;
                }
        }
        break;

        default:
                OSYSLOG1((osyslogERROR, "antEnvMsg::Call: invalid dstSelector (%d)", this->dstSelector));
                return ANT_FAIL;
        }

} // antEnvBaseMsg::Call


antEnvMsg* antEnvMsg::Receive(ANTENVMSG _msg)
{
        return (antEnvMsg*)_msg;
}


int32 antEnvMsg::GetOperation()
{
        return dstOperation;
}


OID& antEnvMsg::GetSender()
{
        return sender;
}


// TCP thread functions

static TCPEndpointError TCP_error(void)
{
        switch (errno) {
                case EADDRINUSE:        return TCP_ADDRESSINUSE;
                case EADDRNOTAVAIL:     return TCP_ADDRESSERROR;
                case ENETUNREACH:       return TCP_NETWORK_UNREACHABLE;
                case EAGAIN:            return TCP_WOULDBLOCK;
#if EAGAIN != EWOULDBLOCK
                case EWOULDBLOCK:       return TCP_WOULDBLOCK;
#endif
                case EFAULT:            return TCP_BUFFER_INVALID;
                default:                return TCP_FAIL;
        }
}


static TCPEndpointError TCP_state_error(OpenSdkEndpointInfo::State state)
{
        switch (state) {
                case OpenSdkEndpointInfo::NEW:          return TCP_CONNECTION_CLOSED;
                case OpenSdkEndpointInfo::CLOSED:       return TCP_CONNECTION_CLOSED;
                case OpenSdkEndpointInfo::LISTENING:    return TCP_CONNECTION_BUSY;
                case OpenSdkEndpointInfo::CONNECTING:   return TCP_CONNECTION_BUSY;
                default:                                return TCP_FAIL;
        }
}


static void* tcpListenThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei       = (OpenSdkEndpointInfo*)arg;
        TCPEndpointListenMsg *listenMsg = (TCPEndpointListenMsg*)osei->msg;
        int listenSocket                = osei->server_socket;

        pthread_mutex_lock(osei->listen_mutex);
        int acceptRet = accept(listenSocket, NULL, NULL);

        if (acceptRet <= 0) {
                listenMsg->error = TCP_error();
        } else {
                osei->socket     = acceptRet;
                osei->state      = OpenSdkEndpointInfo::ACTIVE;
                listenMsg->error = TCP_SUCCESS;
        }

        pthread_mutex_unlock(osei->listen_mutex);

        _sendMessage(osei->entry, listenMsg);

        return NULL;
}


static void* tcpReceiveThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei         = (OpenSdkEndpointInfo*)arg;
        int socket                        = osei->socket;
        int maxSize                       = osei->size;
        void *buffer                      = osei->buffer;
        TCPEndpointReceiveMsg *receiveMsg = (TCPEndpointReceiveMsg*)osei->msg;

        int receiveRet = read(socket, buffer, maxSize);

        if (receiveRet < 0) {
                receiveMsg->error = TCP_error();

        } else if (receiveRet == 0) {
                receiveMsg->error = TCP_CONNECTION_CLOSED;

        } else {
                receiveMsg->sizeMin = receiveRet;
                receiveMsg->sizeMax = receiveRet;
                receiveMsg->error   = TCP_SUCCESS;
        }

        _sendMessage(osei->entry, receiveMsg);

        return NULL;
}


static void* tcpSendThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei   = (OpenSdkEndpointInfo*)arg;
        int socket                  = osei->socket;
        int sendSize                = osei->size;
        void *buffer                = osei->buffer;
        TCPEndpointSendMsg *sendMsg = (TCPEndpointSendMsg*)osei->msg;

        int writeRet = write(socket, buffer, sendSize);

        if (writeRet <= 0) {
                sendMsg->error = TCP_error();
        } else {
                sendMsg->error = TCP_SUCCESS;
        }

        _sendMessage(osei->entry, sendMsg);

        return NULL;
}


static void* tcpConnectThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei         = (OpenSdkEndpointInfo*)arg;
        TCPEndpointConnectMsg *connectMsg = (TCPEndpointConnectMsg*)osei->msg;

        struct sockaddr_in addr;
        memset(&addr, 0, sizeof(addr));

        addr.sin_addr.s_addr = osei->ip.Address();
        addr.sin_family      = AF_INET;
        addr.sin_port        = htons(osei->port);

        int ret = connect(osei->socket, (struct sockaddr *)&addr, sizeof(addr));

        if (ret < 0) {
                connectMsg->error = TCP_error();

        } else {
                connectMsg->error = TCP_SUCCESS;
                osei->state       = OpenSdkEndpointInfo::ACTIVE;
        }

        _sendMessage(osei->entry, connectMsg);

        return NULL;
}


static void* tcpCloseThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei     = (OpenSdkEndpointInfo*)arg;
        TCPEndpointCloseMsg *closeMsg = (TCPEndpointCloseMsg*)osei->msg;

        close(osei->socket);
// can't close the socket below because there might be other threads using it. do we need a reference counting?
//      if (osei->server_socket > 0) close(osei->server_socket);

        osei->socket        = NO_SOCKET;
        osei->server_socket = NO_SOCKET;

        osei->state = OpenSdkEndpointInfo::CLOSED;
        closeMsg->error = TCP_SUCCESS;

        _sendMessage(osei->entry, closeMsg);

        return NULL;
}




static UDPEndpointError UDP_error(void)
{
        switch (errno) {
                case EADDRINUSE:        return UDP_ADDRESSINUSE;
                case EADDRNOTAVAIL:     return UDP_ADDRESSERROR;
                case ENETUNREACH:       return UDP_NETWORK_UNREACHABLE;
                case EAGAIN:            return UDP_WOULDBLOCK;
#if EAGAIN != EWOULDBLOCK
                case EWOULDBLOCK:       return UDP_WOULDBLOCK;
#endif
                case EFAULT:            return UDP_BUFFER_INVALID;
                default:                return UDP_FAIL;
        }
}


static UDPEndpointError UDP_state_error(OpenSdkEndpointInfo::State state)
{
        switch (state) {
                case OpenSdkEndpointInfo::NEW:          return UDP_CONNECTION_CLOSED;
                case OpenSdkEndpointInfo::CLOSED:       return UDP_CONNECTION_CLOSED;
                default:                                return UDP_FAIL;
        }
}


static void* udpReceiveThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei         = (OpenSdkEndpointInfo*)arg;
        int socket                        = osei->socket;
        int recvSize                      = osei->size;
        UDPEndpointReceiveMsg *receiveMsg = (UDPEndpointReceiveMsg*)osei->msg;

        struct sockaddr_in fromAddress;
        socklen_t sizeOfAddr = sizeof(fromAddress);
        memset(&fromAddress, 0, sizeof(fromAddress));

        int receiveRet = recvfrom(socket, osei->buffer, recvSize, 0, (struct sockaddr *) &fromAddress, &sizeOfAddr);

        if (receiveRet <= 0) {
                receiveMsg->error = UDP_error();

        } else {
                receiveMsg->size    = receiveRet;
                receiveMsg->port    = ntohs(fromAddress.sin_port);
                receiveMsg->address = IPAddress((uint32)fromAddress.sin_addr.s_addr);
                receiveMsg->error   = UDP_SUCCESS;
        }

        _sendMessage(osei->entry, receiveMsg);

        delete osei;
        return NULL;
}


static void* udpSendThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei   = (OpenSdkEndpointInfo*)arg;
        UDPEndpointSendMsg *sendMsg = (UDPEndpointSendMsg*)osei->msg;
        int socket                  = osei->socket;
        int sendSize                = osei->size;
        void *buffer                = osei->buffer;

        struct sockaddr_in address;

        memset(&address, 0, sizeof(address));
        address.sin_family      = AF_INET;
        address.sin_port        = htons(sendMsg->port);
        address.sin_addr.s_addr = osei->ip.Address();

        int sendRet = sendto(socket, buffer, sendSize, 0, (struct sockaddr*)&address, sizeof(address));

        if (sendRet <= 0) {
                sendMsg->error = UDP_error();
        } else {
                sendMsg->error = UDP_SUCCESS;
        }

        _sendMessage(osei->entry, sendMsg);

        delete osei;
        return NULL;
}


static void* udpCloseThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei     = (OpenSdkEndpointInfo*)arg;
        UDPEndpointCloseMsg *closeMsg = (UDPEndpointCloseMsg*)osei->msg;

        close(osei->socket);

        osei->socket       = NO_SOCKET;
        osei->state        = OpenSdkEndpointInfo::CLOSED;
        closeMsg->error    = UDP_SUCCESS;

        _sendMessage(osei->entry, closeMsg);

        delete osei;
        return NULL;
}


static DNSEndpointError DNS_error(int error)
{
        switch (error) {
#ifndef __CYGWIN__
                case EAI_AGAIN:         return DNS_TRY_AGAIN;
                case EAI_NODATA:        return DNS_NO_DATA;
#endif
                default:                return DNS_FAIL;
        }
}


static void* dnsGetHostByNameThreadFunc(void *arg)
{
        OpenSdkEndpointInfo *osei        = (OpenSdkEndpointInfo*)arg;
        DNSEndpointGetHostByNameMsg *msg = (DNSEndpointGetHostByNameMsg*)osei->msg;

#ifndef __CYGWIN__
        struct addrinfo *res;
        struct addrinfo hints;
        memset(&hints, 0, sizeof(hints));

        hints.ai_family = AF_INET; // we only support IPv4 for now

        int ret = getaddrinfo(osei->hostname, NULL, &hints, &res);
#else
        int ret = -1;
#endif
        free(osei->hostname);

        if (ret != 0) {
                msg->error     = DNS_error(ret);
                msg->n_address = 0;
                msg->n_alias   = 0;

        } else {
#ifndef __CYGWIN__
                msg->error = DNS_SUCCESS;

                unsigned int count = 0;

                for (struct addrinfo *r = res; r != NULL; r = r->ai_next) {
                        if (!count++) { // use only the first one
                                msg->host_address = IPAddress((uint32)((sockaddr_in*)r->ai_addr)->sin_addr.s_addr);
                        }
                }

                freeaddrinfo(res);

                msg->n_address = count;
                msg->n_alias   = 0;
#endif
        }

        _sendMessage(osei->entry, msg);

        return NULL;
}

---