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/Logger.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 <netinet/tcp.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])
        {
                // 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();
        }

        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(bool errorcheck)
        {
                if ((::close(_socket) == -1) && errorcheck)
                {
                        throw ConnectionClosedException();
                }
        }

        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)
        {
                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)
                {
                        IBRCOMMON_LOGGER_DEBUG(90) << "tcpstream::overflow() nothing to sent" << IBRCOMMON_LOGGER_ENDL;
                        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 ECONNRESET:
                                // Connection reset by peer
                                errmsg = ERROR_RESET;
                                break;

                        default:
                                errmsg = ERROR_WRITE;
                        }

                        // failure
			::close(_socket);
                        std::stringstream ss; ss << "<tcpstream> send() in tcpstream failed: " << errno;
                        throw ConnectionClosedException(ss.str());
                }

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

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

                // end of stream
                if (bytes == 0)
                {
                        errmsg = ERROR_CLOSED;
			::close(_socket);
                        std::stringstream ss; ss << "<tcpstream> recv() returned zero: " << errno;
                        throw ConnectionClosedException(ss.str());
                }
                else if (bytes < 0)
                {
                        switch (errno)
                        {
                        case EPIPE:
                                // connection has been reset
                                errmsg = ERROR_EPIPE;
                                break;

                        default:
                                errmsg = ERROR_READ;
                        }

			::close(_socket);
                        std::stringstream ss; ss << "<tcpstream> recv() failed: " << errno;
                        throw ConnectionClosedException(ss.str());
                }

                // 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");
                }
        }

        void tcpstream::enableLinger(int l)
        {
                // set linger option to the socket
                struct linger linger;

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

        void tcpstream::enableNoDelay()
        {
                int set = 1;
                ::setsockopt(_socket, IPPROTO_TCP, TCP_NODELAY, (char *)&set, sizeof(set));
        }

}