socketstream.hpp

Go to the documentation of this file.

 
#ifndef __JOIN_SOCKETSTREAM_HPP__
#define __JOIN_SOCKETSTREAM_HPP__
 
// libjoin.
#include <join/socket.hpp>
 
// C++.
#include <streambuf>
#include <utility>
#include <memory>
#include <chrono>
 
namespace join
{
    template <class Protocol>
    class BasicSocketStreambuf : public std::streambuf
    {
    public:
        using Endpoint = typename Protocol::Endpoint;
        using Socket   = typename Protocol::Socket;
 
        BasicSocketStreambuf ()
        : _buf (std::make_unique <char []> (2 * _bufsize)),
          _socket (Socket::Mode::NonBlocking)
        {
        }
 
        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)
        {
            this->close ();
 
            std::streambuf::operator= (std::move (other));
            this->_buf = std::move (other._buf);
            this->_timeout = other._timeout;
            this->_socket = std::move (other._socket);
 
            return *this;
        }
 
        virtual ~BasicSocketStreambuf ()
        {
            if (this->_socket.connected ())
            {
                this->overflow (traits_type::eof ());
            }
        }
 
        BasicSocketStreambuf* bind (const Endpoint& endpoint)
        {
            if (this->_socket.bind (endpoint) == -1)
            {
                return nullptr;
            }
 
            return this;
        }
 
        BasicSocketStreambuf* connect (const Endpoint& endpoint)
        {
            if (this->_socket.connect (endpoint) == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return nullptr;
                }
 
                if (!this->_socket.waitConnected (this->_timeout))
                {
                    this->_socket.close ();
                    return nullptr;
                }
            }
 
            return this;
        }
 
        BasicSocketStreambuf* disconnect ()
        {
            if (this->_socket.connected () && (this->overflow (traits_type::eof ()) == traits_type::eof ()))
            {
                return nullptr;
            }
 
            if (this->_socket.disconnect () == -1)
            {
                if (lastError != Errc::TemporaryError)
                {
                    return nullptr;
                }
 
                if (!this->_socket.waitDisconnected (this->_timeout))
                {
                    return nullptr;
                }
            }
 
            return this;
        }
 
        void close ()
        {
            this->_socket.close ();
        }
 
        void timeout (int ms)
        {
            this->_timeout = ms;
        }
 
        int timeout () const
        {
            return this->_timeout;
        }
 
        Socket& socket ()
        {
            return this->_socket;
        }
 
    protected:
        virtual int_type underflow () override
        {
            if (!this->_socket.connected ())
            {
                lastError = make_error_code (Errc::ConnectionClosed);
                return traits_type::eof ();
            }
 
            if (this->eback () == nullptr)
            {
                this->setg (this->_buf.get (), this->_buf.get (), this->_buf.get ());
            }
 
            if (this->gptr () == this->egptr ())
            {
                for (;;)
                {
                    int nread = this->_socket.read (this->eback (), _bufsize);
                    if (nread == -1)
                    {
                        if (lastError == Errc::TemporaryError)
                        {
                            if (this->_socket.waitReadyRead (this->_timeout))
                            {
                                continue;
                            }
                        }
                        this->_socket.close();
                        return traits_type::eof ();
                    }
 
                    this->setg (this->eback (), this->eback (), this->eback () + nread);
                    break;
                }
            }
 
            return traits_type::to_int_type (*this->gptr ());
        }
 
        virtual int_type overflow (int_type c = traits_type::eof ()) override
        {
            if (!this->_socket.connected ())
            {
                lastError = make_error_code (Errc::ConnectionClosed);
                return traits_type::eof ();
            }
 
            if (this->pbase () == nullptr)
            {
                this->setp (this->_buf.get () + _bufsize, this->_buf.get () + (2 * _bufsize));
            }
 
            if ((this->pptr () == this->epptr ()) || (c == traits_type::eof ()))
            {
                std::streamsize pending = this->pptr () - this->pbase ();
                if (pending)
                {
                    if (this->_socket.writeExactly (this->pbase (), pending, this->_timeout) == -1)
                    {
                        this->_socket.close ();
                        return traits_type::eof ();
                    }
                }
 
                this->setp (this->pbase (), this->pbase () + _bufsize);
 
                if (c == traits_type::eof ())
                {
                    return traits_type::not_eof (c);
                }
            }
 
            return this->sputc (traits_type::to_char_type (c));
        }
 
        virtual int_type sync () override
        {
            if (!this->_socket.connected () || (this->overflow () == traits_type::eof ()))
            {
                return -1;
            }
            return 0;
        }
 
        static const std::streamsize _bufsize = 4096;
 
        std::unique_ptr <char []> _buf;
 
        int _timeout = 30000;
 
        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)
        {
        }
 
        BasicSocketStream (const BasicSocketStream& other) = delete;
 
        BasicSocketStream& operator=(const BasicSocketStream& other) = delete;
 
        BasicSocketStream (BasicSocketStream&& other)
        : std::iostream (std::move (other)),
          _sockbuf (std::move (other._sockbuf))
        {
            this->set_rdbuf (&this->_sockbuf);
        }
 
        BasicSocketStream& operator=(BasicSocketStream&& other)
        {
            std::iostream::operator= (std::move (other));
            this->_sockbuf = std::move (other._sockbuf);
            return *this;
        }
 
        virtual ~BasicSocketStream () = default;
 
        virtual void bind (const Endpoint& endpoint)
        {
            if (this->_sockbuf.bind (endpoint) == nullptr)
            {
                this->setstate (std::ios_base::failbit);
            }
        }
 
        virtual void connect (const Endpoint& endpoint)
        {
            if (this->_sockbuf.connect (endpoint) == nullptr)
            {
                this->setstate (std::ios_base::failbit);
            }
        }
 
        virtual void disconnect ()
        {
            if (this->_sockbuf.disconnect () == nullptr)
            {
                this->setstate (std::ios_base::failbit);
            }
        }
 
        virtual void close ()
        {
            this->_sockbuf.close ();
        }
 
        Endpoint localEndpoint ()
        {
            return this->_sockbuf.socket ().localEndpoint ();
        }
 
        const Endpoint& remoteEndpoint ()
        {
            return this->_sockbuf.socket ().remoteEndpoint ();
        }
 
        bool opened ()
        {
            return this->_sockbuf.socket ().opened ();
        }
 
        bool connected ()
        {
            return this->_sockbuf.socket ().connected ();
        }
 
        bool encrypted ()
        {
            return this->_sockbuf.socket ().encrypted ();
        }
 
        void timeout (int ms)
        {
            this->_sockbuf.timeout (ms);
        }
 
        int timeout () const
        {
            return this->_sockbuf.timeout ();
        }
 
        Socket& socket ()
        {
            return this->_sockbuf.socket ();
        }
 
    protected:
        SocketStreambuf _sockbuf;
    };
 
    template <class Protocol>
    class BasicTlsStream : public BasicSocketStream <Protocol>
    {
    public:
        using SocketStreambuf = BasicSocketStreambuf <Protocol>;
        using Endpoint        = typename Protocol::Endpoint;
        using Socket          = typename Protocol::Socket;
 
        BasicTlsStream ()
        : BasicSocketStream <Protocol> ()
        {
        }
 
        BasicTlsStream (const BasicTlsStream& other) = delete;
 
        BasicTlsStream& operator=(const BasicTlsStream& other) = delete;
 
        BasicTlsStream (BasicTlsStream&& other)
        : BasicSocketStream <Protocol> (std::move (other))
        {
        }
 
        BasicTlsStream& operator=(BasicTlsStream&& other)
        {
            BasicSocketStream <Protocol>::operator= (std::move (other));
 
            return *this;
        }
 
        virtual ~BasicTlsStream () = default;
 
        void startEncryption ()
        {
            if (this->_sockbuf.socket ().startEncryption () == -1)
            {
                if (lastError == Errc::TemporaryError)
                {
                    if (this->_sockbuf.socket ().waitEncrypted (this->timeout ()))
                        return;
                }
 
                this->setstate (std::ios_base::failbit);
            }
        }
 
        void connectEncrypted (const Endpoint& endpoint)
        {
            if (this->_sockbuf.connect (endpoint) == nullptr)
            {
                this->setstate (std::ios_base::failbit);
            }
            else
            {
                this->startEncryption ();
                if (this->fail ())
                {
                    this->close ();
                }
            }
        }
 
        int setCertificate (const std::string& cert, const std::string& key = "")
        {
            return this->_sockbuf.socket ().setCertificate (cert, key);
        }
 
        int setCaPath (const std::string& caPath)
        {
            return this->_sockbuf.socket ().setCaPath (caPath);
        }
 
        int setCaFile (const std::string& caFile)
        {
            return this->_sockbuf.socket ().setCaFile (caFile);
        }
 
        void setVerify (bool verify, int depth = -1)
        {
            return this->_sockbuf.socket ().setVerify (verify, depth);
        }
 
        int setCipher (const std::string &cipher)
        {
            return this->_sockbuf.socket ().setCipher (cipher);
        }
 
        int setCipher_1_3 (const std::string &cipher)
        {
            return this->_sockbuf.socket ().setCipher_1_3 (cipher);
        }
    };
}
 
#endif