Class:
==========
MySQLStream:    PackageStream

Interface:
==========
PackageStream

#ifndef THORS_ANVIL_MYSQL_PACKAGE_STREAM_H
#define THORS_ANVIL_MYSQL_PACKAGE_STREAM_H

#include <string>
#include <cstddef>

namespace ThorsAnvil
{
    namespace MySQL
    {

class PackageStream
{
    public:
        virtual ~PackageStream()                                        = 0;
        virtual void        read(char* buffer, std::size_t len)         = 0;
        virtual void        write(char const* buffer, std::size_t len)  = 0;
        virtual void        startNewConversation()                      = 0;
        virtual void        flush()                                     = 0;
        virtual void        reset()                                     = 0;
        virtual void        drop()                                      = 0;
        virtual bool        isEmpty()                                   = 0;
        virtual std::string readRemainingData()                         = 0;
};

    }
}

#endif

#ifndef THORS_ANVIL_MYSQL_MY_SQL_STREAM_H
#define THORS_ANVIL_MYSQL_MY_SQL_STREAM_H

#include "PackageStream.h"
#include <string>
#include <cstddef>

namespace ThorsAnvil
{
    namespace MySQL
    {

class MySQLStream: public PackageStream
{
    static std::size_t constexpr ErrorResult = static_cast<std::size_t>(-1);

    int socket;
    public:
         MySQLStream(std::string const& host, int port);
         MySQLStream(int socket);
        ~MySQLStream();
        virtual void        read(char* buffer, std::size_t len)         override;
        virtual void        write(char const* buffer, std::size_t len)  override;
        virtual void        startNewConversation()                      override {}
        virtual void        flush()                                     override {}
        virtual void        reset()                                     override {}
        virtual void        drop()                                      override {}
        virtual bool        isEmpty()                                   override {return true;}
        virtual std::string readRemainingData()                         override {return "";}
};

    }
}

#endif

#include "MySQLStream.h"
#include "ThorSQL/SQLUtil.h"
#include <stdexcept>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <netdb.h>
#include <unistd.h>
#include <errno.h>

using namespace ThorsAnvil::MySQL;

MySQLStream::MySQLStream(int socket)
    : socket(socket)
{}
MySQLStream::MySQLStream(std::string const& host, int port)
{
    port    = port ? port : 3306;

    sockaddr_in serv_addr;
    memset(&serv_addr, '0', sizeof(serv_addr));
    serv_addr.sin_family    = AF_INET;
    serv_addr.sin_port      = htons(port);

    hostent*    serv  = ::gethostbyname(host.c_str());
    if (serv == NULL)
    {
        throw std::runtime_error(
                errorMsg("ThorsAnvil::MySQL::MySQLStream::MySQLStream: ",
                         "::gethostbyname() Failed: ", strerror(errno)
              ));
    }
    bcopy((char *)serv->h_addr, (char *)&serv_addr.sin_addr.s_addr, serv->h_length);

    if ((socket = ::socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        throw std::runtime_error(
                errorMsg("ThrosAnvil::MySQL::MySQLStream::MySQLStream: ",
                         "::socket() Failed: ", strerror(errno)
              ));
    }

    using SockAddr = struct sockaddr;
    if (::connect(socket, reinterpret_cast<SockAddr*>(&serv_addr), sizeof(serv_addr)) < 0)
    {
        ::close(socket);
        throw std::runtime_error(
                errorMsg("ThorsAnvil::MySQL::MySQLStream::MySQLStream: ",
                         "::connect() Failed: ", strerror(errno)
              ));
    }
}
MySQLStream::~MySQLStream()
{
    ::close(socket);
}


void MySQLStream::read(char* buffer, std::size_t len)
{
    std::size_t     readSoFar    = 0;
    while (readSoFar != len)
    {
        std::size_t read = ::read(socket, buffer + readSoFar, len - readSoFar);
        if ((read == ErrorResult) && (errno == EAGAIN || errno == EINTR))
        {
            /* Recoverable error. Try again. */
            continue;
        }
        else if (read == 0)
        {
            throw std::runtime_error(
                    errorMsg("ThorsAnvil::MySQL::MySQLStream::read: "
                             "::read() Failed: ",
                             "Tried to read ", len, "bytes but only found ", readSoFar, " before EOF"
                  ));
        }
        else if (read == ErrorResult)
        {
            throw std::runtime_error(
                    errorMsg("ThorsAnvil::MySQL::MySQLStream::read: ",
                             "::read() Failed: ",
                             "errno=", errno, " Message=", strerror(errno)
                  ));
        }

        readSoFar += read;
    }
}
void MySQLStream::write(char const* buffer, std::size_t len)
{
    std::size_t     writenSoFar    = 0;
    while (writenSoFar != len)
    {
        std::size_t writen = ::write(socket, buffer + writenSoFar, len - writenSoFar);
        if ((writen == ErrorResult) && (errno == EAGAIN || errno == EINTR))
        {
            /* Recoverable error. Try again. */
            continue;
        }
        else if (writen == 0)
        {
            throw std::runtime_error(
                    errorMsg("ThorsAnvil::MySQL::MySQLStream::write: ",
                             "::write() Failed: ",
                             "Tried to write ", len, "bytes but only found ", writenSoFar, " before EOF"
                  ));
        }
        else if (writen == ErrorResult)
        {
            throw std::runtime_error(
                    errorMsg("ThorsAnvil::MySQL::MySQLStream::write: ",
                             "::write() Failed: ",
                             "errno=", errno, " Message=", strerror(errno)
                  ));
        }

        writenSoFar += writen;
    }
}

#include <sstream>
#include <stdexcept>

class MockStream: public ThorsAnvil::MySQL::PackageStream
{
    char const*   input;
    unsigned char*output;
    std::size_t   len;
    std::size_t   readSoFar;
    std::size_t   writSoFar;
    public:
        MockStream(char const* data, std::size_t len, unsigned char* output = nullptr)
            : input(data)
            , output(output)
            , len(len)
            , readSoFar(0)
            , writSoFar(0)
        {
        }
        virtual void        read(char* buffer, std::size_t size)        override {if (readSoFar + size > len) {
                                                                                    std::stringstream msg;
                                                                                    msg << "Read too much: readSoFar(" << readSoFar << ") Size(" << size << ") Len(" << len << ")";
                                                                                    throw std::runtime_error(msg.str());
                                                                                  }
                                                                                  std::copy(input + readSoFar, input + readSoFar + size, buffer);readSoFar += size;
                                                                                 }
        virtual void        write(char const* buffer, std::size_t len)  override {std::copy(buffer, buffer + len, output + writSoFar); writSoFar += len;}
        virtual void        startNewConversation()                      override {}
        virtual void        flush()                                     override {}
        virtual void        drop()                                      override {readSoFar = len;}
        virtual void        reset()                                     override {}
        virtual bool        isEmpty()                                   override {return len == readSoFar;}
        virtual std::string readRemainingData()                         override {return std::string(input + readSoFar, input + readSoFar + len);}


        std::size_t         readLen() const {return readSoFar;}
        std::size_t         writLen() const {return writSoFar;}
};

class MySQLStream: public PackageStream {
    //...
    virtual void        startNewConversation() override {}
    virtual void        flush()                override {}
    virtual void        reset()                override {}
    virtual void        drop()                 override {}
    virtual bool        isEmpty()              override {return true;}
    virtual std::string readRemainingData()    override {return "";}
    //...
};