hsms_socket.cpp

/*   
 *   (c) Copyright 2008 Philipp Skadorov (philipp_s@users.sourceforge.net)
 *
 *   This file is part of FREESECS.
 *
 *   FREESECS 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, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   FREESECS 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 FREESECS, see COPYING.
 *   If not, see <http://www.gnu.org/licenses/>.
 */
#include <string>
#include <errno.h>
#include <fcntl.h>
#include <sys/param.h>
#include "log.h"
#include "hsms_msg.h"
#include "hsms_socket.h"
#include "tsqueue.h"
#include "hsms_signums.h"
#include "log.h"

#define HSMS_SOCK_LOG_LEVEL 9

    
#define MAXRECV 1024

using namespace freesecs;

hsms_socket_t::hsms_socket_t(const char *name, const int port)
:_remote_host(""), _port(port), _sock(-1)
{
    _name = name;
    _name += " sock";
    _cnx_sm_name = _name;
    _cnx_sm_name += " cnx sm";

    memset (&_addr, 0, sizeof (_addr));

    if(0 != create_socket())
    {
        throw;
    }
}

hsms_socket_t::hsms_socket_t(const char *name, const char *host, const int port)
:_remote_host(host), _port(port), _sock(-1)
{
    _name = name;
    _name += " sock";
    _cnx_sm_name = _name;
    _cnx_sm_name += " cnx sm";

    memset (&_addr, 0, sizeof (_addr));

    if(0 != create_socket())
    {
        throw;
    }
}

int
hsms_socket_t::create_socket()
{

    _sock = socket (AF_INET, SOCK_STREAM, 0);
    if (_sock <= 0)
    {
        TRACE_ERROR("%s: failed creating streaming socket", __func__);
        return errno;
    }
    
    int on = 1;

    if (setsockopt (_sock, 
                SOL_SOCKET, 
                SO_REUSEADDR, 
                (const char *) &on, 
                sizeof (on)) == -1)
    {
        TRACE_ERROR("%s: failed setting socket options", __func__);
        return errno;
    }

    return 0;
}

hsms_socket_t::~hsms_socket_t () 
{
    disconnect();
}

int
hsms_socket_t::bind () 
{
    if (_sock <= 0)
    {
        return EBADF;
    }
    _addr.sin_family = AF_INET;
    _addr.sin_addr.s_addr = htonl (INADDR_ANY);
    _addr.sin_port = htons (_port);
    
    if(-1 == ::bind(_sock, (struct sockaddr*)&_addr, sizeof(_addr)))
    {
        return errno;
    }

    return 0;
}

int
hsms_socket_t::listen () 
{
    if (_sock <= 0)
    {
        return EBADF;
    }

    if(-1 == ::listen (_sock, MAXCONNECTIONS))
    {
        return errno;
    }

    return 0;
}

int
hsms_socket_t::accept () 
{
    sock_cnx_ev_t ev;
    int new_sock = 0;
    int addr_length = sizeof (_addr);

    new_sock =::accept (_sock, (sockaddr *) & _addr, (socklen_t *) & addr_length);
    
    if (new_sock <= 0)
    {
        return errno;
    }
    TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "hsms_socket_t::accept: cnx established");

    ::shutdown(_sock, SHUT_RDWR);
    ::close (_sock);
    _sock = new_sock;

    return 0;
}

int freesecs::hsms_socket_t::connect_to() 
{
    if (_sock <= 0)
    {
        TRACE_ERROR("hsms_socket_t::remote_connect - socket is not valid");
        return EBADF;
    }
    _addr.sin_family = AF_INET;
    _addr.sin_port = htons (_port);
    int
    status = inet_pton (AF_INET, _remote_host.c_str (), &_addr.sin_addr);

    if (status && errno == EAFNOSUPPORT)
    {
        TRACE_ERROR("hsms_socket_t::remote_connect - invalid address: %s", _remote_host.c_str());
        return errno;
    }
    
    if(-1 == ::connect (_sock, (sockaddr *) & _addr, sizeof (_addr)))
    {
        TRACE_ERROR("hsms_socket_t::remote_connect - failed connecting to address: %s/%d", 
                                                                    _remote_host.c_str(), _port);
        return errno;
    }
    TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "hsms_socket_t::remote_connect to %s/%d OK!", 
                                                                _remote_host.c_str(),  _port);

    return 0;
}

int
hsms_socket_t::connect_tcp()
{
    int r = 0;

    if(0 == _sock)
    {
        if(0 != (r=create_socket()))
        {
            TRACE_ERROR("hsms_socket_t::connect: create socket failed: %s", strerror(r));
            return r;
        }
    }

    if(_remote_host.length() == 0)
    {
        TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "hsms_socket_t::connect: bind");
        if(0 != (r = bind()))
        {
            TRACE_ERROR("hsms_socket_t::connect: bind failed: %s", strerror(errno));
            return r;
        }
        if(0 != (r = listen()))
        {
            TRACE_ERROR("hsms_socket_t::connect: listen failed: %s", strerror(errno));
            return r;
        }

        TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL,"hsms_socket_t::connect: accept");

        r = accept();
        if(0 != r)
        {
            TRACE_ERROR("hsms_socket_t::connect: accept failed: %s", strerror(errno));
            return r;
        }
    }
    else
    {
        r = connect_to();
    }
    return r;
}

int
hsms_socket_t::disconnect_tcp()
{
    int r = 0;

    if(::shutdown(_sock, SHUT_RDWR))
    {
        r = errno;
        TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "hsms_socket_t::disconnect: shutdown returned %s", strerror(r));
    }

    if(::close(_sock))
    {
        r = errno;
        TRACE_ERROR("hsms_socket_t::disconnect: close returned %s", strerror(r));
    }

    _sock = 0;

    //return r;
    return 0;
}

hsms_socket_t::cnx_state_et 
hsms_socket_t::get_cnx_state() const
{
    return _cnx_state;
}

int
hsms_socket_t::connect()
{
    sock_cnx_ev_t ev = {e_connect,};
    return process(ev);
}

int
hsms_socket_t::disconnect()
{
    sock_cnx_ev_t ev = {e_disconnect,};
    return process(ev);
}

int 
hsms_socket_t::send (pdata_t pdata)
{
    sock_cnx_ev_t ev = {e_send, pdata};
    return process(ev);
}

int 
hsms_socket_t::receive(pdata_t pdata)
{
    sock_cnx_ev_t ev = {e_recv, pdata};
    return process(ev);
}

int
hsms_socket_t::send_tcp(pdata_t pd)
{
    int r = 0;
    ssize_t sent, size = (*pd).size();
    
    sent = ::send (_sock, (void*)pd->data(), size, MSG_NOSIGNAL);

    if(sent != size)
    {
        r = errno;
    }

    return r;
}

int
hsms_socket_t::process(sock_cnx_ev_t ev)
{
    int r = 0;
    cnx_state_et state_old = _cnx_state;

    TRACE_FSM_BEGIN(HSMS_SOCK_LOG_LEVEL, _cnx_sm_name.c_str(), _cnx_state, ev.type);

    switch(_cnx_state)
    {
        case NOT_CONNECTED:
        if(e_connect == ev.type)
        {
            _cnx_state = CONNECTING;
        }
        else
        {
            goto sm_wrong_ev;
        }
        break;
        case CONNECTING:
        if(e_connected == ev.type)
        {
            _cnx_state = CONNECTED;
        }
        else if(e_connect_failed == ev.type)
        {
            r = z3(ev);
            _cnx_state = NOT_CONNECTED;
        }
        else if(e_disconnect == ev.type)
        {
            _cnx_state = NOT_CONNECTED;
        }
        else
        {
            goto sm_wrong_ev;
        }
        break;
        case CONNECTED:
        if(e_disconnect == ev.type)
        {
            _cnx_state = NOT_CONNECTED;
        }
        else if(e_send == ev.type)
        {
            r = z1(ev);
        }
        else if(e_recv == ev.type)
        {
            r = z2(ev);
        }
        else
        {
            goto sm_wrong_ev;
        }
        break;
        default:
            r = ENOTSUP;
        break;
    }

    if(state_old == _cnx_state)
    {
        goto sm_end;
    }

    TRACE_FSM_TRANS(HSMS_SOCK_LOG_LEVEL, _cnx_sm_name.c_str(), state_old, _cnx_state);
    
    switch(_cnx_state)
    {
        case NOT_CONNECTED:
            r = z1_0(ev);
        break;
        case CONNECTING:
            r = z2_0(ev);
        break;
        case CONNECTED:
            r = z3_0(ev);
        break;
        default:
            r = ENOTSUP;
        break;
    }
sm_end:
    TRACE_FSM_END(HSMS_SOCK_LOG_LEVEL, _cnx_sm_name.c_str(), _cnx_state, ev.type);
    return r;
sm_wrong_ev:

    TRACE_FSM_ERROR(HSMS_SOCK_LOG_LEVEL, _cnx_sm_name.c_str(),
                    ", %s: event %d not handled in current state: %d", 
                    _name.c_str(), ev.type, _cnx_state);
    TRACE_FSM_END(HSMS_SOCK_LOG_LEVEL, _name.c_str(), _cnx_state, ev.type);
    return ENOTSUP;
}

int 
hsms_socket_t::z1(sock_cnx_ev_t ev)
{
    return send_tcp(ev.pdata);
}

int 
hsms_socket_t::z2(sock_cnx_ev_t ev)
{
    return recv_tcp(ev.pdata);
}

int
hsms_socket_t::z3(sock_cnx_ev_t)
{
    cnx_failed_signal();
    return 0;
}

int 
hsms_socket_t::z1_0(sock_cnx_ev_t ev)
{
    int r = disconnect_tcp();
    cnx_state_signal(_cnx_state);
    return r;
}

int 
hsms_socket_t::z2_0(sock_cnx_ev_t ev)
{
    int r = 0;
    pthread_t tid;
    pthread_attr_t attr;


    cnx_state_signal(_cnx_state);

    pthread_attr_init(&attr);
    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);

    if(pthread_create(&tid, &attr, tcp_cnx_thread, (void*)this))
    {
        r = errno;
    }

    pthread_attr_destroy(&attr);

    return r;
}

int 
hsms_socket_t::z3_0(sock_cnx_ev_t ev)
{
    cnx_state_signal(_cnx_state);
    return 0;
}

void *
hsms_socket_t::tcp_cnx_thread(void *arg)
{
    int r;
    sock_cnx_ev_t ev;
    hsms_socket_t *psock = (hsms_socket_t*)arg;

    r = psock->connect_tcp();
    if(0 == r)
    {
        ev.type = e_connected;
    }
    else
    {
        ev.type = e_connect_failed;
    }

    psock->process(ev);

    pthread_exit(NULL);
}

/*<<<async socket>>>*/
hsms_socket_async_t::hsms_socket_async_t(const char *name, const int port)
:hsms_socket_t(name, port)
{
}

hsms_socket_async_t::hsms_socket_async_t(const char *name, const char *host, int port)
:hsms_socket_t(name, host, port),_ar(NULL)
{

}

hsms_socket_async_t::~hsms_socket_async_t()
{
    disconnect();
}

int
hsms_socket_async_t::disconnect_tcp()
{
    process(e_i_deinit);
    return freesecs::hsms_socket_t::disconnect_tcp();
}

int 
hsms_socket_async_t::recv_tcp(pdata_t pmessage)
{
    _pmessage = pmessage;
    process(e_i_init);
    return EWOULDBLOCK;
}

#if 0
bool 
hsms_socket_async_t::x1()
{
    bool valid = HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN == _bytes_read;
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid;

}

bool 
hsms_socket_async_t::x2()
{
    int msg_len = ntohl(HSMS_MSG_LEN(&(*_pmessage)[0]));
    bool valid = msg_len >= HSMS_HDR_LEN;
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": len %d, valid %d", 
                            msg_len, 
                            valid);
    return valid;
}

bool 
hsms_socket_async_t::x3()
{
/*    hsms_msg_base<data_t> msg_base;
    bool valid = msg_base.valid(*_pmessage);
*/
    bool valid = HSMS_HDR_LEN <= ntohl(HSMS_MSG_LEN(&(*_pmessage)[0]))
        && _pmessage->size() >= HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN
            && hsms_msg_base<data_t>::e_separate_req 
                                   >= HSMS_MSG_STYPE(&(*_pmessage)[0]);

    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid;
}

bool 
hsms_socket_async_t::x4()
{
    hsms_msg_base<data_t> msg_base;
    bool valid = hsms_msg_base<data_t>::e_data == msg_base.type(*_pmessage);
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid;
}

bool 
hsms_socket_async_t::x5()
{
    bool valid = _rx_error > 0;
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid;
}

bool 
hsms_socket_async_t::x6()
{
    data_msg<data_t> msg;
    bool valid = msg.text_len(*_pmessage) <= _bytes_read
                    - HSMS_LEN_BYTES_LEN - HSMS_HDR_LEN;
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid; 
}

int 
hsms_socket_async_t::z0()
{
    int len = HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN;

    if(!_pmessage.get())
    {
        _pmessage = new data_t;
    }
    //_pmessage->clear();
    _pmessage->resize(len, 0);

    if(NULL == _ar.get())
    {
        _ar = new async_reception_t(_sock, _name.c_str());
        _ar->data_partially_recvd_signal.add_handler(this, &freesecs::hsms_socket_async_t::data_rcvd_handler);
        _ar->data_recvd_signal.add_handler(this, &freesecs::hsms_socket_async_t::data_rcvd_handler);
        _ar->recv_error_signal.add_handler(this, &freesecs::hsms_socket_async_t::rcv_error_handler);
    }
    _ar->start_recv((void*)&(*_pmessage)[0], len);

    return 0;
}

int 
hsms_socket_async_t::z1()
{
    data_msg<data_t> msg;
    
    unsigned text_len = msg.text_len(*_pmessage);
    unsigned total_len = HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN + text_len;
    
    TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL,
                "%s::z1: text length %d,"
                " total length %d", 
                _name.c_str(),
                text_len, total_len);

    if(_bytes_read < total_len)
    {
        _pmessage->resize(total_len, 0);
    }

    _ar->start_recv((void*)&(*_pmessage)[HSMS_DATABYTES_OFFSET], text_len);

    return 0;
}

int 
hsms_socket_async_t::z3()
{
    state_signal(e_again);
    return 0;
}

int 
hsms_socket_async_t::z4()
{
    state_signal(e_again);
    return 0;
}

int 
hsms_socket_async_t::z5()
{
    TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "%s::z5: disconnect...", _name.c_str());

    _ar->stop_recv();
    _ar = NULL;
    return 0;
}

int 
hsms_socket_async_t::z1_0()
{
    _rx_error = 0;
    _bytes_read = 0;
    *_pmessage = data_t(HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN, 0);

    state_signal(_state);
    return 0;
}

int 
hsms_socket_async_t::z2_0()
{
    state_signal(_state);
    return 0;
}

int 
hsms_socket_async_t::z3_0()
{
    state_signal(_state);
    msg_signal(_pmessage);
    return 0;
}

int 
hsms_socket_async_t::z4_0()
{
    state_signal(_state);
    return 0;
}
#endif

void 
hsms_socket_async_t::process(internal_event_t e)
{
    state_t state_old = _state;

    TRACE_FSM_BEGIN(HSMS_SOCK_LOG_LEVEL, _name.c_str(), _state, e);

    switch(_state)
    {
        case inactive:
            if(e_i_init == e)
            {
                _state = wtg_for_header;
            }
        break;
        case wtg_for_header:
            if(e_i_deinit == e)
            {
                _state = inactive;
            }
            else if(e_i_data_received == e)
            {
                //if the whole header received,
                //go to wtg_for_data state
                if(x0())
                {
                    if(x2())
                    {
                        z1();
                        z1_0();
                    }
                    else
                    {
                        _state = wtg_for_data;
                    }
                }
                else
                {
                    z2();
                }
            }
            else if(e_i_recv_error == e)
            {
               //just reset the data buf
               //do not change the state
               z0();
               z3();
            }
        break;
        case wtg_for_data:
            if(e_i_deinit == e)
            {
                _state = inactive;
            }
            else if(e_i_data_received == e)
            {
                //if the whole msg received,
                //transit to wtg_for_header state
                if(x1())
                {
                    //signal msg recieved
                    z1();
                    _state = wtg_for_header;
                }
                else
                {
                    z2();
                }
            }
            else if(e_i_recv_error == e)
            {
                //reset the databuf and transit
                //to wtg_for_header state
                z3();
                _state = wtg_for_header;
            }
        default:
        break;
    }

    if(state_old == _state)
    {
        goto sm_end;
    }

    TRACE_FSM_TRANS(HSMS_SOCK_LOG_LEVEL, _name.c_str(), state_old, _state);

    switch(_state)
    {
        case inactive:
            z0_0();
        break;
        case wtg_for_header:
            z1_0();
        break;
        case wtg_for_data:
            z2_0();
        default:
        break;
    }

sm_end:
    TRACE_FSM_END(HSMS_SOCK_LOG_LEVEL, _name.c_str(), _state, e);
    return;
}

bool 
hsms_socket_async_t::x0()
{
    bool valid = HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN == _bytes_read;
    valid &= HSMS_HDR_LEN <= ntohl(HSMS_MSG_LEN(&(*_pmessage)[0]))
        && _pmessage->size() >= HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN
            && hsms_msg_base<data_t>::e_separate_req 
                                   >= HSMS_MSG_STYPE(&(*_pmessage)[0]);
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid;
}
bool 
hsms_socket_async_t::x1()
{
    hsms_msg_base<data_t> msg_base;

    bool valid = msg_base.valid(*_pmessage);
    TRACE_FSM_ACTION(HSMS_SOCK_LOG_LEVEL, _name.c_str(), __func__, ": %d", valid);
    return valid;
}
bool 
hsms_socket_async_t::x2()
{
    return HSMS_HDR_LEN == ntohl(HSMS_MSG_LEN(&(*_pmessage)[0]));
}
void 
hsms_socket_async_t::z0()
{
    _rx_error = 0;
    _bytes_read = 0;
    *_pmessage = data_t(HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN, 0);
}
void 
hsms_socket_async_t::z1()
{
    msg_signal(_pmessage);
}
void 
hsms_socket_async_t::z2()
{
    data_partially_recvd();
}
void 
hsms_socket_async_t::z3()
{
    recv_error();
}
void 
hsms_socket_async_t::z0_0()
{
    _ar->stop_recv();
    _ar = NULL;
}
void 
hsms_socket_async_t::z1_0()
{
    _rx_error = 0;
    _bytes_read = 0;
    *_pmessage = data_t(HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN, 0);

    if(NULL == _ar.get())
    {
        _ar = new async_reception_t(_sock, _name.c_str());
        _ar->data_partially_recvd_signal.add_handler(this, &freesecs::hsms_socket_async_t::data_rcvd_handler);
        _ar->data_recvd_signal.add_handler(this, &freesecs::hsms_socket_async_t::data_rcvd_handler);
        _ar->recv_error_signal.add_handler(this, &freesecs::hsms_socket_async_t::rcv_error_handler);
    }

    _ar->start_recv(&(*_pmessage)[0], _pmessage->size());
}
void 
hsms_socket_async_t::z2_0()
{
    data_msg<data_t> msg;
    
    unsigned text_len = msg.text_len(*_pmessage);
    unsigned total_len = HSMS_LEN_BYTES_LEN + HSMS_HDR_LEN + text_len;
    
    TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL,
                "%s::z1: text length %d,"
                " total length %d", 
                _name.c_str(),
                text_len, total_len);

    if(_bytes_read < total_len)
    {
        _pmessage->resize(total_len, 0);
    }

    _ar->start_recv((void*)&(*_pmessage)[HSMS_DATABYTES_OFFSET], text_len);
}

#if 0
void 
hsms_socket_async_t::process(internal_event_t e)
{
    state_t state_old = _state;

    TRACE_FSM_BEGIN(HSMS_SOCK_LOG_LEVEL, _name.c_str(), _state, e);

    switch(_state)
    {
        case idle:
            if(e_i_init == e)
            {
                z0();
                _state = wtg_for_header;
            }
        break;
        case msg_received:
        case error:
            if(e_i_init == e)
            {
                z0();
                _state = wtg_for_header;
            }
            else if(e_i_cancelled == e)
            {
                z5();
                _state = idle;
            }
        break;
        case wtg_for_header:
            if(e_i_data_received == e)
            {
                if(x5())
                {
                    _state = error;
                }
                else if(!x1())
                {
                    z3();
                }
                else if(!(x2() && x3()))
                {
                    _state = error;
                }
                else if(x4())
                {
                    z1();
                    _state = wtg_for_data;
                }
                else
                {
                    _state = msg_received;
                }
            }
            else if(e_i_cancelled == e)
            {
                z5();
                _state = idle;
            }
#ifdef GRACEFUL_RECEIVE
            if(error == _state)
            {
                z0();
                state_old = idle;
                _state= wtg_for_header;
            }
#endif
        break;
        case wtg_for_data:
            if(e_i_data_received == e)
            {
                if(!x6())
                {
                    z4();
                }
                else if(!x5())
                {
                    _state = msg_received;
                }
                else
                {
                    _state = error;
                }

            }
            else if(e_i_cancelled == e)
            {
                z5();
                _state = idle;
            }
        default:
        break;
    }

    if(state_old == _state)
    {
        goto sm_end;
    }

    TRACE_FSM_TRANS(HSMS_SOCK_LOG_LEVEL, _name.c_str(), state_old, _state);

    switch(_state)
    {
        case idle:
        break;
        case wtg_for_header:
            z1_0();
        break;
        case wtg_for_data:
            z2_0();
        break;
        case msg_received:
            z3_0();
        break;
        case error:
            z4_0();
        break;
        default:
        break;
    }

sm_end:
    TRACE_FSM_END(HSMS_SOCK_LOG_LEVEL, _name.c_str(), _state, e);
    return;
}
#endif

int 
hsms_socket_async_t::data_rcvd_handler(const size_t &num_bytes)
{
    _bytes_read += num_bytes;
    process(e_i_data_received);
    return 0;
}

int 
hsms_socket_async_t::rcv_error_handler(const int &err)
{
    _rx_error = err;
    if(ECANCELED == err)
    {
        process(e_i_deinit);
    }
    else
    {
        _bytes_read = 0;
        process(e_i_data_received);
    }
    return 0;
}


/*<<<<timed socket>>>>*/

#ifdef USE_TIMED_SOCK
static char
    buf[MAXRECV];
int
hsms_socket_timed_t::_recv (data_t & data, size_t size, 
                            bool wait_all, bool flash_unwanted, 
                            int timeout) const
{
    int retcode = 0;

    if (MAXRECV < size)
    return -1;
    int flags = wait_all && size ? MSG_WAITALL : 0;

    size_t actual_size = flash_unwanted ? MAXRECV : size;
    memset (buf, 0, MAXRECV);
    int status =::recv (_sock, buf, actual_size, flags);

    if (status < 0)
    
    {
        TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "hsms_socket_timed_t::_recv: status: -1   errno: %d", errno);
        if (EWOULDBLOCK == errno)
        
        {
            TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "recv timeout in hsms_socket_timed_t::recv!!");
            retcode = -1;
        }
        retcode = status;
    }
    
    else if (status == 0)
    
    {
        TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "cnx closed from other side!");
        retcode = -2;
    }
    
    else if (size > 0)
    
    {
        char *begin = buf;
        char *end = buf + status;

        data.reserve (status);
        data.assign (begin, end);
    }
    return retcode;
}
int
hsms_socket_timed_t::recv (data_t & d, int timeout_ms) const
{
    if(timeout_ms)
    {
        _set_timer (_sock, timeout_ms);
    }
    return _recv (d, MAXRECV, false, false, 0);
}
int
hsms_socket_timed_t::recvn (data_t & d, size_t size, int timeout_ms) const 
{
    if(timeout_ms)
    {
        _set_timer (_sock, timeout_ms);
    }
    return _recv (d, size, true, false, 0);
}

void
hsms_socket_timed_t::set_non_blocking (const bool b) 
{
    int opts;

    opts = fcntl (_sock, F_GETFL);
    if (opts < 0)
    
    {
        return;
    }
    if (b)
    opts = (opts | O_NONBLOCK);
    
    else
    opts = (opts & ~O_NONBLOCK);
    fcntl (_sock, F_SETFL, opts);
}
void
hsms_socket_timed_t::_set_timer (int sock_fd, int timeout_ms) 
{
    timeval tv;

    if (sock_fd <= 0)
    {
        return;
    }

    if (0 >= timeout_ms)
    {
        tv.tv_sec = 0;
        tv.tv_usec = 0;
    }
    
    else
    
    {
        tv.tv_sec = timeout_ms / 1000;
        tv.tv_usec = MAX (0, MAX (0, timeout_ms - 1000 * tv.tv_sec) * 1000);
    }
    ::setsockopt (sock_fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof (tv));
    TRACE_DEBUG(HSMS_SOCK_LOG_LEVEL, "sock timer: sec=%d usec=%d", tv.tv_sec, tv.tv_usec);
}
#endif/*USE_TIMED_SOCK*/

---