Have a working version of MySQL implementation of ThorsSQL library done.
If you want to check it out you can find the whole thing on github ThorsSQL.
This is a follow on to previous code Reviews:
Part 3: Layer 1
Part 2
Part 1
The documentation for these classes is here:
Part 3 (Layer 2): The MySQL Implementation
MySQL uses a package based protocol. Each request/response message is wrapped in a set of packages. These packages are numbered and sized to allow validation and some basic error checking of communication between the client and server. The class PackageBuffer implements the PackageStream interface; but it is used to wrap a Layer 1 stream thus hiding the packages from the next layer and still exposing a simple stream like interface.
Class:
==========
PackageBuffer: PackageStream
Interface:
==========
PackageStream
It should be noted that the implementation of this layer is currently very basic. One of the improvement points is implementing a more efficient version of this package stream.
PackageBuffer.h
#ifndef THORS_ANVIL_MYSQL_PACKAGE_BUFFER_H
#define THORS_ANVIL_MYSQL_PACKAGE_BUFFER_H
#include "PackageStream.h"
#include <vector>
#include <string>
#include <cstddef>
namespace ThorsAnvil
{
namespace MySQL
{
template<typename T>
class PackageBuffer: public PackageStream
{
T& stream;
std::size_t readCurrentPacketSize;
std::size_t readCurrentPacketPosition;
unsigned char currentPacketSequenceID;
bool hasMore;
bool flushed;
std::vector<char> sendBuffer;
private:
void nextPacket();
void writePackageHeader(std::size_t size);
void writeStream(char const* buffer, std::size_t len);
public:
PackageBuffer(T& stream, bool flushed = false);
virtual void read(char* buffer, std::size_t len) override;
virtual void write(char const* buffer, std::size_t len) override;
virtual bool isEmpty() override;
virtual void startNewConversation() override;
virtual void flush() override;
virtual void drop() override;
virtual void reset() override;
virtual std::string readRemainingData() override;
};
}
}
#ifndef COVERAGE_MySQL
#include "PackageBuffer.tpp"
#endif
#endif
PackageBuffer.tpp
#include "ThorSQL/SQLUtil.h"
#include <iomanip>
namespace ThorsAnvil
{
namespace MySQL
{
template<typename T>
PackageBuffer<T>::PackageBuffer(T& stream, bool flushed)
: stream(stream)
, readCurrentPacketSize(0)
, readCurrentPacketPosition(0)
, currentPacketSequenceID(-1)
, hasMore(true)
, flushed(flushed)
{
sendBuffer.reserve(0xFFFFFF/*16MByte*/);
}
template<typename T>
inline void PackageBuffer<T>::read(char* buffer, std::size_t len)
{
std::size_t retrieved = 0;
do
{
std::size_t remaining = readCurrentPacketSize - readCurrentPacketPosition;
std::size_t getFromPacket = std::min((len - retrieved), remaining);
stream.read(buffer + retrieved, getFromPacket);
retrieved += getFromPacket;
readCurrentPacketPosition += getFromPacket;
if (retrieved != len && remaining == 0)
{ nextPacket();
}
}
while (len != retrieved);
}
template<typename T>
bool PackageBuffer<T>::isEmpty()
{
long remaining = readCurrentPacketSize - readCurrentPacketPosition;
if ((remaining == 0) && hasMore)
{
nextPacket();
remaining = readCurrentPacketSize - readCurrentPacketPosition;
}
return remaining == 0;
}
template<typename T>
std::string PackageBuffer<T>::readRemainingData()
{
std::string dst;
do
{
long retrieved = dst.size();
long remaining = readCurrentPacketSize - readCurrentPacketPosition;
dst.resize(retrieved + remaining);
read(&dst[retrieved], remaining);
if (hasMore)
{
nextPacket();
continue;
}
break;
}
while (true);
return dst;
}
template<typename T>
void PackageBuffer<T>::nextPacket()
{
if (!hasMore)
{
throw std::domain_error(
errorMsg("ThorsAnvil::MySQL::PackageBuffer::nextPacket: ",
"No more data expected from server"
));
}
readCurrentPacketSize = 0;
readCurrentPacketPosition = 0;
currentPacketSequenceID++;
std::uint32_t packetBufferSize = 0;
// TODO FIX only works on little endian
stream.read(reinterpret_cast<char*>(&packetBufferSize), 3);
readCurrentPacketSize = packetBufferSize;
char actualSequenceID;
stream.read(&actualSequenceID, 1);
if (currentPacketSequenceID != actualSequenceID)
{
throw std::domain_error(
errorMsg("ThorsAnvil::MySQL::PackageBuffer::nextPacket: ",
"currentPacketSequenceID(", currentPacketSequenceID, ")",
" != actual sequence on input stream(", actualSequenceID, ")"
));
}
hasMore = readCurrentPacketSize == 0xFFFFFF;
}
template<typename T>
void PackageBuffer<T>::startNewConversation()
{
currentPacketSequenceID = -1;
}
template<typename T>
void PackageBuffer<T>::write(char const* buffer, std::size_t len)
{
if (flushed)
{
throw std::domain_error(
bugReport("ThorsAnvil::MySQL::PackageBuffer::write: ",
"Writting to a flushed buffer"
));
}
std::size_t currentSize = sendBuffer.size();
if (currentSize + len >= 0xFFFFFF)
{
std::size_t available = 0xFFFFFF - currentSize;
writePackageHeader(0xFFFFFF);
writeStream(&sendBuffer[0], currentSize);
writeStream(buffer, available);
sendBuffer.clear();
buffer += available;
len -= available;
while (len > 0xFFFFFF)
{
writePackageHeader(0xFFFFFF);
writeStream(buffer, 0xFFFFFF);
buffer += 0xFFFFFF;
len -= 0xFFFFFF;
}
}
sendBuffer.insert(sendBuffer.end(), buffer, buffer + len);
}
template<typename T>
void PackageBuffer<T>::flush()
{
if (flushed)
{
throw std::domain_error(
bugReport("ThorsAnvil::MySQL::PackageBuffer<T>::flush: ",
"Already flushed"
));
}
flushed = true;
std::size_t currentSize = sendBuffer.size();
writePackageHeader(currentSize);
writeStream(&sendBuffer[0], currentSize);
sendBuffer.clear();
}
template<typename T>
void PackageBuffer<T>::drop()
{
std::size_t dataLeft;
std::vector<char> drop;
do
{
std::size_t readDataAvailable = readCurrentPacketSize - readCurrentPacketPosition;
drop.resize(readDataAvailable);
read(&drop[0], readDataAvailable);
dataLeft = 0;
if (hasMore)
{
nextPacket();
dataLeft = readCurrentPacketSize - readCurrentPacketPosition;
}
}
while (dataLeft != 0);
}
template<typename T>
void PackageBuffer<T>::reset()
{
std::size_t readDataAvailable = readCurrentPacketSize - readCurrentPacketPosition;
if (readDataAvailable == 0 && hasMore)
{
nextPacket();
readDataAvailable = readCurrentPacketSize - readCurrentPacketPosition;
}
if (readDataAvailable != 0)
{
std::stringstream extraData;
for (std::size_t loop=0; loop < readDataAvailable; ++loop)
{
char x;
read(&x, 1);
extraData << std::hex << std::setw(2) << std::setfill('0') << static_cast<int>(x) << "(" << x << ") ";
}
throw std::domain_error(
bugReport("ThorsAnvil::MySQL::PackageBuffer<T>::reset: ",
"reset() before message was read:",
extraData.str()
));
}
flushed = false;
hasMore = true; // Will allow us to start reading the next packet
}
template<typename T>
void PackageBuffer<T>::writePackageHeader(std::size_t size)
{
++currentPacketSequenceID;
stream.write(reinterpret_cast<char const*>(&size), 3);
stream.write(reinterpret_cast<char const*>(¤tPacketSequenceID), 1);
}
template<typename T>
void PackageBuffer<T>::writeStream(char const* buffer, std::size_t len)
{
stream.write(buffer, len);
}
}
}
