socket_stream.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_SOCKET_STREAM_HPP
#define JOIN_CORE_SOCKET_STREAM_HPP
 
// libjoin.
#include <join/stream_socket.hpp>
 
// C++.
#include <streambuf>
#include <utility>
#include <chrono>
#include <memory>
 
namespace join
{
    template <class Protocol>
    class BasicSocketStreambuf : public std::streambuf
    {
    public:
        using Endpoint = typename Protocol::Endpoint;
        using Socket = typename Protocol::Socket;
        using TimePoint = typename Socket::TimePoint;
 
        BasicSocketStreambuf ()
        : _buf (std::make_unique<char[]> (2 * _bufsize))
        , _socket (Socket::Mode::NonBlocking)
        {
        }
 
        explicit BasicSocketStreambuf (Socket&& socket)
        : _buf (std::make_unique<char[]> (2 * _bufsize))
        , _socket (std::move (socket))
        {
        }
 
        BasicSocketStreambuf (const BasicSocketStreambuf& other) = delete;
 
        BasicSocketStreambuf& operator= (const BasicSocketStreambuf& other) = delete;
 
        BasicSocketStreambuf (BasicSocketStreambuf&& other)
        : std::streambuf (std::move (other))
        , _buf (std::move (other._buf))
        , _timeout (other._timeout)
        , _socket (std::move (other._socket))
        {
        }
 
        BasicSocketStreambuf& operator= (BasicSocketStreambuf&& other)
        {
            close ();
 
            std::streambuf::operator= (std::move (other));
            _buf = std::move (other._buf);
            _timeout = other._timeout;
            _socket = std::move (other._socket);
 
            return *this;
        }
 
        virtual ~BasicSocketStreambuf ()
        {
            if (_socket.connected ())
            {
                overflow (traits_type::eof ());
            }
        }
 
        BasicSocketStreambuf* bind (const Endpoint& endpoint)
        {
            if (_socket.bind (endpoint) == -1)
            {
                return nullptr;
            }
 
            return this;
        }
 
        BasicSocketStreambuf* connect (const Endpoint& endpoint)
        {
            if (_socket.connect (endpoint) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return nullptr;
                }
 
                if (!_socket.waitConnected (deadline ()))
                {
                    _socket.close ();
                    return nullptr;
                }
            }
 
            return this;
        }
 
        BasicSocketStreambuf* disconnect ()
        {
            if (_socket.connected () && (overflow (traits_type::eof ()) == traits_type::eof ()))
            {
                return nullptr;
            }
 
            if (_socket.disconnect () == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return nullptr;
                }
 
                if (!_socket.waitDisconnected (deadline ()))
                {
                    return nullptr;
                }
            }
 
            return this;
        }
 
        void close ()
        {
            _socket.close ();
        }
 
        void timeout (std::chrono::nanoseconds timeout)
        {
            _timeout = timeout;
        }
 
        std::chrono::nanoseconds timeout () const
        {
            return _timeout;
        }
 
        TimePoint deadline () const noexcept
        {
            return (_timeout == std::chrono::nanoseconds::zero ()) ? TimePoint::max ()
                                                                   : std::chrono::steady_clock::now () + _timeout;
        }
 
        Socket& socket ()
        {
            return _socket;
        }
 
    protected:
        virtual int_type underflow () override
        {
            if (!_socket.connected ())
            {
                lastError = make_error_code (Errc::ConnectionClosed);
                return traits_type::eof ();
            }
 
            if (eback () == nullptr)
            {
                setg (_buf.get (), _buf.get (), _buf.get ());
            }
 
            if (gptr () == egptr ())
            {
                for (;;)
                {
                    ssize_t nread = _socket.read (eback (), _bufsize);
                    if (nread == -1)
                    {
                        if (lastError == Errc::TemporaryError)
                        {
                            if (_socket.waitReadyRead (deadline ()))
                            {
                                continue;
                            }
                        }
                        _socket.close ();
                        return traits_type::eof ();
                    }
 
                    setg (eback (), eback (), eback () + nread);
                    break;
                }
            }
 
            return traits_type::to_int_type (*gptr ());
        }
 
        virtual int_type overflow (int_type c = traits_type::eof ()) override
        {
            if (!_socket.connected ())
            {
                lastError = make_error_code (Errc::ConnectionClosed);
                return traits_type::eof ();
            }
 
            if (pbase () == nullptr)
            {
                setp (_buf.get () + _bufsize, _buf.get () + (2 * _bufsize));
            }
 
            if ((pptr () == epptr ()) || (c == traits_type::eof ()))
            {
                std::streamsize pending = pptr () - pbase ();
                if (pending)
                {
                    if (_socket.writeExactly (pbase (), pending, deadline ()) == -1)
                    {
                        _socket.close ();
                        return traits_type::eof ();
                    }
                }
 
                setp (pbase (), pbase () + _bufsize);
 
                if (c == traits_type::eof ())
                {
                    return traits_type::not_eof (c);
                }
            }
 
            return sputc (traits_type::to_char_type (c));
        }
 
        virtual int_type sync () override
        {
            if (!_socket.connected () || (overflow () == traits_type::eof ()))
            {
                return -1;
            }
            return 0;
        }
 
        static const std::streamsize _bufsize = 4096;
 
        std::unique_ptr<char[]> _buf;
 
        std::chrono::nanoseconds _timeout = std::chrono::seconds (30);
 
        Socket _socket;
    };
 
    template <class Protocol>
    class BasicSocketStream : public std::iostream
    {
    public:
        using SocketStreambuf = BasicSocketStreambuf<Protocol>;
        using Endpoint = typename Protocol::Endpoint;
        using Socket = typename Protocol::Socket;
 
        BasicSocketStream ()
        : std::iostream (&_sockbuf)
        {
        }
 
        explicit BasicSocketStream (Socket&& socket)
        : std::iostream (&_sockbuf)
        , _sockbuf (std::move (socket))
        {
        }
 
        BasicSocketStream (const BasicSocketStream& other) = delete;
 
        BasicSocketStream& operator= (const BasicSocketStream& other) = delete;
 
        BasicSocketStream (BasicSocketStream&& other)
        : std::iostream (std::move (other))
        , _sockbuf (std::move (other._sockbuf))
        {
            set_rdbuf (&_sockbuf);
        }
 
        BasicSocketStream& operator= (BasicSocketStream&& other)
        {
            std::iostream::operator= (std::move (other));
            _sockbuf = std::move (other._sockbuf);
            return *this;
        }
 
        virtual ~BasicSocketStream () = default;
 
        virtual void bind (const Endpoint& endpoint)
        {
            if (_sockbuf.bind (endpoint) == nullptr)
            {
                setstate (std::ios_base::failbit);
            }
        }
 
        virtual void connect (const Endpoint& endpoint)
        {
            if (_sockbuf.connect (endpoint) == nullptr)
            {
                setstate (std::ios_base::failbit);
            }
        }
 
        virtual void disconnect ()
        {
            if (_sockbuf.disconnect () == nullptr)
            {
                setstate (std::ios_base::failbit);
            }
        }
 
        virtual void close ()
        {
            _sockbuf.close ();
        }
 
        Endpoint localEndpoint ()
        {
            return _sockbuf.socket ().localEndpoint ();
        }
 
        Endpoint remoteEndpoint ()
        {
            return _sockbuf.socket ().remoteEndpoint ();
        }
 
        bool opened ()
        {
            return _sockbuf.socket ().opened ();
        }
 
        bool connected ()
        {
            return _sockbuf.socket ().connected ();
        }
 
        void timeout (std::chrono::nanoseconds timeout)
        {
            _sockbuf.timeout (timeout);
        }
 
        std::chrono::nanoseconds timeout () const
        {
            return _sockbuf.timeout ();
        }
 
        Socket& socket ()
        {
            return _sockbuf.socket ();
        }
 
    protected:
        SocketStreambuf _sockbuf;
    };
}
 
#endif