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 2
Part 3: Layer 1
Part 2
Part 1
The documentation for these classes is here:
Part 3 (Layer 3): The MySQL Implementation
This layer knows how to encode/de-code specific objects onto a stream. At this layer we have two classes ConnectWriter/ConectReader. These classes have methods to encode/de-code each specific type understood by the MySQL server onto a PackageStream. Objects of these type are constructed using a PackageStream and write/read data from this stream.
ConectReader.h
#ifndef THORS_ANVIL_MYSQL_PACKAGE_READER_H
#define THORS_ANVIL_MYSQL_PACKAGE_READER_H
#include "MySQLTimeBag.h"
#include "ThorSQL/SQLUtil.h"
#include <memory>
#include <functional>
#include <vector>
#include <map>
#include <string>
#include <ctime>
#include "MySQLConfig.h"
#if defined(THOR_ENDIAN_SML)
#define THOR_MYSQL_READ_INT(into, len) stream.read(reinterpret_cast<char*>(&into), len)
#elif defined(THOR_ENDIAN_BIG)
#error Have not defined this for large endian systems.
#else
#error Unknow Endianess
#endif
namespace ThorsAnvil
{
namespace MySQL
{
class PackageStream;
class RespPackage;
class RespPackageEOF;
class ConectReader
{
PackageStream& stream;
unsigned long capbil;
unsigned long charset;
unsigned long long lengthEncodedIntegerUsingSize(unsigned char size);
public:
using OKAction = std::function<RespPackage*(int byte, ConectReader&)>;
using OKMap = std::map<int, OKAction>;
ConectReader(PackageStream& stream)
: stream(stream)
, capbil(0)
, charset(0)
{}
void initFromHandshake(unsigned long capabilities, unsigned long charset);
std::unique_ptr<RespPackageEOF> recvMessageEOF();
std::unique_ptr<RespPackage> recvMessage(OKMap const& actions = {});
public:
void read(char* data, std::size_t len);
bool isEmpty() const;
unsigned long getCapabilities() {return capbil;}
template<int len>
unsigned long long fixedLengthInteger(unsigned long rCap) {return (rCap & capbil) ? fixedLengthInteger<len>() : 0;}
unsigned long long lengthEncodedInteger(unsigned long rCap) {return (rCap & capbil) ? lengthEncodedInteger() : 0;}
std::string fixedLengthString(std::size_t s, unsigned long rCap) {return (rCap & capbil) ? fixedLengthString(s) : "";}
std::string nulTerminatedString(unsigned long rCap) {return (rCap & capbil) ? nulTerminatedString() : "";}
std::string variableLengthString(std::size_t s, unsigned long rCap) {return (rCap & capbil) ? variableLengthString(s) : "";}
std::string lengthEncodedString(unsigned long rCap) {return (rCap & capbil) ? lengthEncodedString() : "";}
std::string restOfPacketString(unsigned long rCap) {return (rCap & capbil) ? restOfPacketString() : "";}
template<int len>
unsigned long long fixedLengthInteger();
unsigned long long lengthEncodedInteger();
std::string fixedLengthString(std::size_t size);
std::string nulTerminatedString();
std::string variableLengthString(std::size_t size);
std::string lengthEncodedString();
std::string restOfPacketString();
std::vector<char> lengthEncodedBlob();
std::time_t readDate();
std::time_t readRel();
unsigned long long readRelMicro();
MySQLTimeBag readDateIntoTimeBag();
MySQLTimeBag readTimeIntoTimeBag();
void drop();
void reset();
};
}
}
#ifndef COVERAGE_MySQL
#include "ConectReader.tpp"
#endif
#endif
ConectWriter.h
#ifndef THORS_ANVIL_MYSQL_PACKAGE_WRITER_H
#define THORS_ANVIL_MYSQL_PACKAGE_WRITER_H
#include <string>
#include <vector>
#include <ctime>
#include "MySQLConfig.h"
#if defined(THOR_ENDIAN_SML)
#define THOR_MYSQL_WRITE_INT(from, len) stream.write(reinterpret_cast<char const*>(&from), len)
#elif defined(THOR_ENDIAN_BIG)
#error Have not defined this for large endian systems.
#else
#error Unknow Endianess
#endif
namespace ThorsAnvil
{
namespace MySQL
{
class PackageStream;
class ConectWriter
{
protected:
PackageStream& stream;
unsigned long capabilities;
unsigned long charset;
public:
ConectWriter(PackageStream& stream)
: stream(stream)
, capabilities(0)
, charset(0)
{}
virtual ~ConectWriter() {}
void initFromHandshake(unsigned long capabilities, unsigned long charset);
template<int len>
void writeFixedLengthInteger(unsigned long long value);
void writeLengthEncodedInteger(unsigned long long value);
void writeFixedLengthString(std::string const& value, std::size_t size); // Makes sure the string is correct size
void writeNullTerminatedString(std::string const& value); // Adds NULL terminator
void writeVariableLengthString(std::string const& value); // Not NULL terminated.
void writeLengthEncodedString(std::string const& value);
void writeRawData(char const* buffer, std::size_t size);
void writeLengthEncodedBlob(std::vector<char> const& value);
void writeDate(std::time_t const& value);
void writeRel(std::time_t const& value);
void writeRel(unsigned long long value);
void flush();
void reset();
};
}
}
#ifndef COVERAGE_MySQL
#include "ConectWriter.tpp"
#endif
#endif
ConectReader.tpp
#include "PackageStream.h"
namespace ThorsAnvil
{
namespace MySQL
{
template<int len>
inline unsigned long long ConectReader::fixedLengthInteger()
{
unsigned long long result = 0;
THOR_MYSQL_READ_INT(result, len);
return result;
}
}
}
ConectWriter.tpp
#include "PackageStream.h"
namespace ThorsAnvil
{
namespace MySQL
{
template<int len>
void ConectWriter::writeFixedLengthInteger(unsigned long long value)
{
THOR_MYSQL_WRITE_INT(value, len);
}
}
}
ConectReader.cpp
#include "PackageStream.h"
#include "ConectReader.h"
#include "RespPackageOK.h"
#include "RespPackageEOF.h"
#include "RespPackageERR.h"
using namespace ThorsAnvil::MySQL;
void ConectReader::initFromHandshake(unsigned long newCapabilities, unsigned long newCharset)
{
capbil = newCapabilities;
charset = newCharset;
}
void ConectReader::read(char* data, std::size_t len)
{
stream.read(data, len);
}
bool ConectReader::isEmpty() const
{
return stream.isEmpty();
}
std::unique_ptr<RespPackageEOF> ConectReader::recvMessageEOF()
{
std::unique_ptr<RespPackage> result = recvMessage({{0xFE, [](int firstByte, ConectReader& reader){return new RespPackageEOF(firstByte, reader);}}});
return downcastUniquePtr<RespPackageEOF>(std::move(result));
}
std::unique_ptr<RespPackage> ConectReader::recvMessage(OKMap const& actions /*= {}*/)
{
int packageType = fixedLengthInteger<1>();;
auto find = actions.find(packageType);
if (find != actions.end())
{
return std::unique_ptr<RespPackage>(find->second(packageType, *this));
}
else if (packageType == 0x00)
{
return std::unique_ptr<RespPackage>(new RespPackageOK(packageType, *this));
}
else if (packageType == 0xFE)
{
// EOF default action: => read and ignore.
RespPackageEOF eofPackage(packageType, *this);
return nullptr;
}
else if (packageType == 0xFF)
{
// Error default action: => read and throw
RespPackageERR errorPackage(packageType, *this);
throw std::runtime_error(
errorMsg("ThorsAnvil::MySQL::ConectReader::recvMessage: ", "Error Message from Server: ", errorPackage.message()
));
}
else
{
find = actions.find(-1);
if (find != actions.end())
{
return std::unique_ptr<RespPackage>(find->second(packageType, *this));
}
throw std::domain_error(
errorMsg("ThorsAnvil::MySQL::ConectReader::recvMessage: ", "Unknown Result Type: ", packageType));
}
}
unsigned long long ConectReader::lengthEncodedInteger()
{
unsigned char type;
read(reinterpret_cast<char*>(&type), 1);
return lengthEncodedIntegerUsingSize(type);
}
unsigned long long ConectReader::lengthEncodedIntegerUsingSize(unsigned char type)
{
unsigned long result;
switch (type)
{
case 0xFA:
case 0xFB:
case 0xFF:
throw std::domain_error(
errorMsg("ThorsAnvil::MySQL::ConectReader::lengthEncodedInteger: ", "Invalid length encoding: ", type));
case 0xFC: result = fixedLengthInteger<2>(); break;
case 0xFD: result = fixedLengthInteger<3>(); break;
case 0xFE: result = fixedLengthInteger<8>(); break;
default: result = type;
}
return result;
}
std::string ConectReader::fixedLengthString(std::size_t size)
{
std::string result(size, ' ');
read(&result[0], size);
return result;
}
std::string ConectReader::nulTerminatedString()
{
std::string result;
char x;
for (read(&x, 1); x != '\0'; read(&x, 1))
{
result.append(1, x);
}
return result;
}
std::string ConectReader::variableLengthString(std::size_t size)
{
return fixedLengthString(size);
}
std::string ConectReader::lengthEncodedString()
{
unsigned long size = lengthEncodedInteger();
std::string result(size, '\0');
read(&result[0], size);
return result;
}
std::string ConectReader::restOfPacketString()
{
return stream.readRemainingData();
}
std::vector<char> ConectReader::lengthEncodedBlob()
{
long size = lengthEncodedInteger();
std::vector<char> result(size, '\0');
read(&result[0], size);
return result;
}
std::time_t ConectReader::readDate()
{
MySQLTimeBag timeBag = readDateIntoTimeBag();
tm time;
time.tm_sec = timeBag.second;
time.tm_min = timeBag.minute;
time.tm_hour = timeBag.hour;
time.tm_mday = timeBag.day;
time.tm_mon = timeBag.month - 1;
time.tm_year = timeBag.year - 1900;
time.tm_isdst = false;
time.tm_zone = NULL;
time.tm_gmtoff = 0;
return timegm(&time);
}
MySQLTimeBag ConectReader::readDateIntoTimeBag()
{
MySQLTimeBag timeBag;
long size = fixedLengthInteger<1>();
if (size != 11 && size != 7 && size != 4 && size != 0)
{
throw std::domain_error(
errorMsg("ThorsAnvil::MySQL::ConectReader::readDate: ", "Invalid Date Size", size, "\nExpecting: 11/7/4/0"));
}
if (size == 11 || size == 7 || size == 4)
{
timeBag.year = fixedLengthInteger<2>();
timeBag.month = fixedLengthInteger<1>();
timeBag.day = fixedLengthInteger<1>();
}
if (size == 11 || size == 7)
{
timeBag.hour = fixedLengthInteger<1>();
timeBag.minute = fixedLengthInteger<1>();
timeBag.second = fixedLengthInteger<1>();
}
if (size == 11)
{
timeBag.uSecond = fixedLengthInteger<4>();
}
return timeBag;
}
std::time_t ConectReader::readRel()
{
MySQLTimeBag timeBag = readTimeIntoTimeBag();
return timeBag.day * (60LL*60*24) + timeBag.hour * (60LL*60) + timeBag.minute * (60LL) + timeBag.second;
}
unsigned long long ConectReader::readRelMicro()
{
MySQLTimeBag timeBag = readTimeIntoTimeBag();
return timeBag.day * (1000LL*60*60*24)
+ timeBag.hour * (1000LL*60*60)
+ timeBag.minute * (1000LL*60)
+ timeBag.second * (1000LL)
+ timeBag.uSecond;
}
MySQLTimeBag ConectReader::readTimeIntoTimeBag()
{
MySQLTimeBag timeBag;
long size = fixedLengthInteger<1>();
if (size != 12 && size != 8 && size != 0)
{
throw std::domain_error(
errorMsg("ThorsAnvil::MySQL::ConectReader::readTime: ", "Invalid Time Size: ", size, "\nExpecting 12/8/0"));
}
timeBag.type = MySQLTimeBag::RelativePositive;
if (size == 12 || size == 8)
{
long negativeTest = fixedLengthInteger<1>();
if (negativeTest < 0 || negativeTest > 1)
{
throw std::domain_error( errorMsg("ThorsAnvil::MySQL::ConectReader::readTime: ", "Invalid Negative Test"));
}
if (negativeTest == 1)
{
timeBag.type = MySQLTimeBag::RelativeNegative;
}
timeBag.day = fixedLengthInteger<4>();
timeBag.hour = fixedLengthInteger<1>();
timeBag.minute = fixedLengthInteger<1>();
timeBag.second = fixedLengthInteger<1>();
}
if (size == 12)
{
timeBag.uSecond = fixedLengthInteger<4>();
}
return timeBag;
}
void ConectReader::drop()
{
stream.drop();
}
void ConectReader::reset()
{
stream.reset();
}
ConectWriter.cpp
#include "PackageStream.h"
#include "ConectWriter.h"
using namespace ThorsAnvil::MySQL;
void ConectWriter::initFromHandshake(unsigned long newCapabilities, unsigned long newCharset)
{
capabilities = newCapabilities;
charset = newCharset;
}
void ConectWriter::writeLengthEncodedInteger(unsigned long long value)
{
/*
* Length encoded integers.
* Val < 251 => 1bytes
* Val >= 251 < 2^16 => 0Xfc + 2byte Value
* Val >= 2^16 < 2^24 => 0Xfd + 3byte Value
* Val >= 2^24 < 2^64 => 0Xfe + 8byte Value
*/
static char const mark2Byte = '\xFC';
static char const mark3Byte = '\xFD';
static char const mark8Byte = '\xFE';
if (value < 251) { writeFixedLengthInteger<1>(value);}
else if (value <= 0xFFFF) { stream.write(&mark2Byte, 1); writeFixedLengthInteger<2>(value);}
else if (value <= 0xFFFFFF) { stream.write(&mark3Byte, 1); writeFixedLengthInteger<3>(value);}
else { stream.write(&mark8Byte, 1); writeFixedLengthInteger<8>(value);}
}
void ConectWriter::writeFixedLengthString(std::string const& value, std::size_t size)
{
std::string output(value);
output.resize(size);
writeVariableLengthString(output);
}
void ConectWriter::writeNullTerminatedString(std::string const& value)
{
stream.write(value.c_str(), value.size() + 1);
}
void ConectWriter::writeVariableLengthString(std::string const& value)
{
stream.write(&value[0], value.size());
}
void ConectWriter::writeLengthEncodedString(std::string const& value)
{
writeLengthEncodedInteger(value.size());
writeVariableLengthString(value);
}
void ConectWriter::writeRawData(char const* buffer, std::size_t size)
{
stream.write(buffer, size);
}
void ConectWriter::writeLengthEncodedBlob(std::vector<char> const& value)
{
writeLengthEncodedInteger(value.size());
stream.write(&value[0], value.size());
}
void ConectWriter::writeDate(std::time_t const& )
{
}
void ConectWriter::writeRel(std::time_t const& )
{
}
void ConectWriter::writeRel(unsigned long long )
{
}
void ConectWriter::flush()
{
stream.flush();
}
void ConectWriter::reset()
{
stream.startNewConversation();
}
