ibrcommon/ibrcommon/net/tcpstream.cpp

Go to the documentation of this file.

/*
 * tcpstream.cpp
 *
 *  Created on: 29.07.2009
 *      Author: morgenro
 */

#include "ibrcommon/config.h"
#include "ibrcommon/net/tcpstream.h"
#include "ibrcommon/thread/MutexLock.h"
#include <netinet/in.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <errno.h>

namespace ibrcommon
{
tcpstream::tcpstream(int socket) :
        iostream(this), errmsg(ERROR_NONE), _socket(socket), in_buf_(
                        new char[BUFF_SIZE]), out_buf_(new char[BUFF_SIZE]), _state(
                        TCPSTREAM_CONNECTED)
{
        // set linger option to the socket
        struct linger linger;

        linger.l_onoff = 1;
        linger.l_linger = 0;
        setsockopt(_socket, SOL_SOCKET, SO_LINGER, &linger, sizeof(linger));

        // Initialize get pointer.  This should be zero so that underflow is called upon first read.
        setg(0, 0, 0);
        setp(out_buf_, out_buf_ + BUFF_SIZE - 1);
}

tcpstream::~tcpstream()
{
        delete[] in_buf_;
        delete[] out_buf_;

        // finally, close the socket
	::close(_socket);
}

string tcpstream::getAddress() const
{
        struct ::sockaddr_in sa;
        int iLen = sizeof(sa);

        getpeername(_socket, (sockaddr*) &sa, (socklen_t*) &iLen);
        return inet_ntoa(sa.sin_addr);
}

int tcpstream::getPort() const
{
        struct ::sockaddr_in sa;
        int iLen = sizeof(sa);

        getpeername(_socket, (sockaddr*) &sa, (socklen_t*) &iLen);
        return ntohs(sa.sin_port);
}

void tcpstream::close()
{
        if (_state == TCPSTREAM_HALF_CLOSED)
        {
                ::shutdown(_socket, SHUT_RDWR);
                _state = TCPSTREAM_CLOSED;
        }
        else if (::close(_socket) == -1)
        {
                _state = TCPSTREAM_CLOSED;
                throw ConnectionClosedException();
        }

        _state = TCPSTREAM_CLOSED;
}

void tcpstream::done()
{
        flush();
        _state = TCPSTREAM_HALF_CLOSED;
        ::shutdown(_socket, SHUT_WR);
}

int tcpstream::sync()
{
        int ret = std::char_traits<char>::eq_int_type(this->overflow(
                        std::char_traits<char>::eof()), std::char_traits<char>::eof()) ? -1
                        : 0;

        return ret;
}

int tcpstream::overflow(int c)
{
        if (_state >= TCPSTREAM_HALF_CLOSED)
        {
                return std::char_traits<char>::eof();
        }

        char *ibegin = out_buf_;
        char *iend = pptr();

        // mark the buffer as free
        setp(out_buf_, out_buf_ + BUFF_SIZE - 1);

        if (!std::char_traits<char>::eq_int_type(c, std::char_traits<char>::eof()))
        {
                *iend++ = std::char_traits<char>::to_char_type(c);
        }

        // if there is nothing to send, just return
        if ((iend - ibegin) == 0)
        {
                return std::char_traits<char>::not_eof(c);
        }

        // send the data
        if (::send(_socket, out_buf_, (iend - ibegin), MSG_NOSIGNAL) < 0)
        {
                switch (errno)
                {
                case EPIPE:
                        // connection has been reset
                        errmsg = ERROR_EPIPE;
                        break;

                case 104:
                        // Connection reset by peer
                        errmsg = ERROR_RESET;
                        break;

                default:
                        errmsg = ERROR_WRITE;
                }

                // failure
                ::shutdown(_socket, SHUT_WR);
                _state = TCPSTREAM_HALF_CLOSED;
                return std::char_traits<char>::eof();
        }

        return std::char_traits<char>::not_eof(c);
}

int tcpstream::underflow()
{
        if (_state == TCPSTREAM_CLOSED)
        {
                return std::char_traits<char>::eof();
        }

        int bytes = ::recv(_socket, in_buf_, BUFF_SIZE, 0);

        // end of stream
        if (bytes == 0)
        {
                errmsg = ERROR_CLOSED;
                ::shutdown(_socket, SHUT_RD);
                _state = TCPSTREAM_CLOSED;
                return std::char_traits<char>::eof();
        }
        else if (bytes < 0)
        {
                switch (errno)
                {
                case EPIPE:
                        // connection has been reset
                        errmsg = ERROR_EPIPE;
                        break;

                default:
                        errmsg = ERROR_READ;
                }

                ::shutdown(_socket, SHUT_RD);
                _state = TCPSTREAM_CLOSED;
                return std::char_traits<char>::eof();
        }

        // Since the input buffer content is now valid (or is new)
        // the get pointer should be initialized (or reset).
        setg(in_buf_, in_buf_, in_buf_ + bytes);

        return std::char_traits<char>::not_eof(in_buf_[0]);
}

void tcpstream::enableKeepalive()
{
        /* Set the option active */
        int optval = 1;
        if(setsockopt(_socket, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(optval)) < 0) {
                throw ibrcommon::SocketException("tcpstream: can not activate keepalives");
        }
}

}

---