Refactored part of the original code:
This review is looking only at the "ConnectionBufferMongo" class. This is derived from "ThorsSocket::SocketStreamBuffer" which is an implementation of std::streambuf that works with TCP/IP sockets. This provides some protected helper methods:
- writeToStream(): Write to underlying stream (track written amount)
- getSocket(): Gets the underlying socket so you can read without tracking.
- getSocket().getMessageData(): Blocking read.
- getSocket().tryGetMessageData(): Non Blocking read.
- incrementInCount(): Increment the tracking for read operations.
- reserveOutputSize(): Makes sure the output buffer has at least n bytes
- reserveInputSize(): Makes sure the input buffer has at least n bytes
Note: This code assume little endian. There is a check (not shown here) that generates a build error if the machine is not little endian. This simplifies the code a lot (sorry for people on big endian machines).
I wanted to write this derived version of the socket buffer so it can take into account things that are specific to Mongo. So it is written knowing that the messages on the stream are Mongo message objects (and thus have specific headers etc).
As a result it handles a couple of things on the fly:
- Compression / DeCompression is handled by the buffer.
The user of the stream can assume that all messages on the stream are OP_MSG. Any OP_COPRESSED messages are converted. Other messages generate an error. - Generating the CRC32-C checksum flag is done automatically (if specified in the header).
- Writes are buffered up and nothing is sent until the whole message to mongo is known (this makes compression easy).
- Reading will try and read a whole message in a single read. BUT will not block (unless it needs to) thus allowing the code to processes partial messages as they become available.
MongoUtil.h
#ifndef THORSANVIL_DB_MONGO_MONGO_UTIL_H
#define THORSANVIL_DB_MONGO_MONGO_UTIL_H
#include <cstdint>
namespace ThorsAnvil::DB::Mongo
{
enum class OP_MsgFlag: std::uint32_t
{
empty = 0,
checksumPresent = 1 << 0, // The message ends with 4 bytes containing a CRC-32C [1] checksum. See Checksum for details.
moreToCome = 1 << 1, // Another message will follow this one without further action from the receiver.
// The receiver MUST NOT send another message until receiving one with moreToCome set to 0 as
// sends may block, causing deadlock. Requests with the moreToCome bit set will not receive a reply.
// Replies will only have this set in response to requests with the exhaustAllowed bit set.
exhaustAllowed = 1 << 16, // The client is prepared for multiple replies to this request using the moreToCome bit.
// The server will never produce replies with the moreToCome bit set unless the request has this bit set.
};
inline std::int32_t operator&(OP_MsgFlag lhs, OP_MsgFlag rhs)
{
return static_cast<std::uint32_t>(lhs) & static_cast<std::uint32_t>(rhs);
}
enum class OpCode : std::uint32_t
{
OP_NOOP = 0,
OP_COMPRESSED = 2012, // Handled specially (see Op_MSgHeader.cpp)
OP_MSG = 2013, // Send a message using the format introduced in MongoDB 3.6.
};
enum class Compression : std::uint8_t
{
Snappy = 1
};
struct OpMsgHeader
{
// Message Header.
std::uint32_t messageSize;
std::uint32_t messageId;
std::uint32_t messageResponseId;
OpCode opCode;
};
struct OpMsgBlock
{
// Message Header.
std::uint32_t messageSize;
std::uint32_t messageId;
std::uint32_t messageResponseId;
OpCode opCode;
// OP_MSG body
OP_MsgFlag flags;
std::uint8_t kind;
// BSON Object
};
struct OpCompressedBlock
{
// Message Header.
std::uint32_t messageSize;
std::uint32_t messageId;
std::uint32_t messageResponseId;
OpCode opCode;
// OP_COMPRESSED Message Body
OpCode originalOpCode;
std::uint32_t uncompressedSize;
Compression compression;
// Compressed Data
};
static constexpr std::uint32_t kSizeUInt32 = 4;
static constexpr std::uint32_t kSizeOpMsgHeaderSize = 16;
static constexpr std::uint32_t kSizeOpMsgBlock = 21;
static constexpr std::uint32_t kSizeOpCompressedBlock = 25;
static constexpr std::uint32_t kSizeOpCompressedMessage = 9;
}
#endif
ConnectionBufferMongo.h
#ifndef THORSANVIL_DB_CONNECTION_BUFFER_MONGO_H
#define THORSANVIL_DB_CONNECTION_BUFFER_MONGO_H
#include "ThorsMongoConfig.h"
#include "MongoUtil.h"
#include "ThorsSocket/SocketStreamBuffer.h"
#include "ThorsCrypto/crc.h"
namespace ThorsAnvil::DB::Mongo
{
class ConnectionBufferMongo: public ThorsSocket::SocketStreamBuffer
{
using traits = std::streambuf::traits_type;
using int_type = traits::int_type;
using char_type = traits::char_type;
using Checksum = ThorsAnvil::Crypto::CRC32C_Checksum;
std::uint32_t inMessageSize = 0;
std::uint32_t outMessageSize = 0;
std::uint32_t underflowNeeds = 1;
std::uint32_t outMessagePlaced = 0; // Used to make sure we get the bytes indicating message size.
bool useCompressionOnWrite = false;
public:
using SocketStreamBuffer::SocketStreamBuffer;
ConnectionBufferMongo(ConnectionBufferMongo&& move) noexcept;
void setCompressionOnWrite(bool comp = true) {useCompressionOnWrite = comp;}
protected:
virtual int_type underflow() override;
virtual std::streamsize xsgetn(char_type* dest, std::streamsize count) override;
virtual int_type overflow(int_type ch = traits::eof()) override;
virtual std::streamsize xsputn(char_type const* source, std::streamsize count) override;
virtual int sync() override;
private:
std::uint32_t messageRetrievedSize() const {return egptr() - eback();}
std::uint32_t messageReadSize() const {return gptr() - eback();}
std::uint32_t missingData() const {return inMessageSize - messageRetrievedSize();}
std::uint32_t availableData() const {return egptr() - gptr();}
};
}
#endif
ConnectionBufferMongo.cpp
#include "ConnectionBufferMongo.h"
#include "snappy.h"
using namespace ThorsAnvil::DB::Mongo;
ConnectionBufferMongo::ConnectionBufferMongo(ConnectionBufferMongo&& move) noexcept
: SocketStreamBuffer(std::move(move))
, inMessageSize(std::exchange(move.inMessageSize, 0))
, outMessageSize(std::exchange(move.outMessageSize, 0))
, underflowNeeds(std::exchange(move.underflowNeeds, 0))
, outMessagePlaced(std::exchange(move.outMessagePlaced, 0))
{}
ConnectionBufferMongo::int_type ConnectionBufferMongo::underflow()
{
/*
* Ensures that at least one character is available in the input area by updating the pointers
* to the input area (if needed) * and reading more data in from the input sequence
* (if applicable).
*
* Returns the value of that character (converted to int_type with Traits::to_int_type(c)) on success
* or Traits::eof() on failure.
*
* The function may update gptr, egptr and eback pointers to define the location of newly
* loaded data (if any).
*
* On failure, the function ensures that either gptr() == nullptr or gptr() == egptr.
* The base class version of the function does nothing. The derived classes may override this function
* to allow updates to the get area in the case of exhaustion.
*/
if (gptr() == egptr())
{
std::size_t newDataCount = 0;
// Check to see if we are in the middle of reading a message.
if ((gptr() - eback()) == inMessageSize)
{
// So we are reading a new message.
// ================================
// We are about to reset the eback(), gptr(), egptr()
// The "SocketStreamBuffer" requires we record the amount of data that was read so
// we can keep track of the amount of the input position correctly.
incrementInCount(inMessageSize);
// Set the buffer back to empty.
setg(eback(), eback(), eback());
inMessageSize = 0;
// The message header is 4 std::uint32_t objects.
// Force the read of the whole header block this will give us size information in the
// first 4 bytes and the OpCode in the last 4 bytes.
// If the OpCode is OP_Compress then we need to read the full message to decompress.
ThorsSocket::IOData result = getSocket().getMessageData(egptr(), kSizeOpMsgHeaderSize);
newDataCount += result.dataSize;
if (newDataCount != kSizeOpMsgHeaderSize)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "underflow", "Failed to get next message header: Got: ", newDataCount);
return traits::eof();
}
// Yes I know this looks risky.
// Alignment comment.
// The underlying streambuffer guarantees that the beginning of the read buffer "eback()"
// is aligned to alignof(std::max_align_t). This means that the buffer is aligned for
// all alignment types.
// Details: See: 6.7.6 Alignment [basic.align]
OpMsgHeader* headerInfo = reinterpret_cast<OpMsgHeader*>(eback());
if (headerInfo->opCode == OpCode::OP_COMPRESSED)
{
getSocket().getMessageData(egptr() + kSizeOpMsgHeaderSize, kSizeOpCompressedMessage);
// See: "Alignment comment" above
OpCompressedBlock* message = reinterpret_cast<OpCompressedBlock*>(eback());
if (message->compression != Compression::Snappy)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "underflow", "Failed to uncompress. Only support snappy compression. Compression: ", static_cast<unsigned int>(messag
return traits::eof();
}
// Retrieve the compressed data into a separate buffer.
std::uint32_t compressMessageSize = message->messageSize - kSizeOpCompressedBlock;
std::vector<char> compressedData;
compressedData.resize(compressMessageSize);
getSocket().getMessageData(&compressedData[0], compressMessageSize);
// Fix the message header with the uncompressed info.
message->messageSize = kSizeOpMsgHeaderSize + message->uncompressedSize;
message->opCode = message->originalOpCode;
reserveInputSize(message->messageSize);
// Add the uncompressed data to the underlying buffer.
bool suc = snappy::RawUncompress(&compressedData[0], compressMessageSize, eback() + kSizeOpMsgHeaderSize);
if (!suc)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "underflow", "Failed to uncompress data with snappy");
return traits::eof();
}
inMessageSize = message->messageSize;
newDataCount = inMessageSize;
}
else if (headerInfo->opCode == OpCode::OP_MSG)
{
inMessageSize = headerInfo->messageSize;
// Make sure the buffer has enough space for the whole message.
reserveInputSize(inMessageSize);
}
else
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "underflow", "Unknown message opCode: ", static_cast<unsigned int>(headerInfo->opCode));
return traits::eof();
}
setg(eback(), gptr(), egptr() + newDataCount);
}
std::size_t extraData = 0;
// See: "Alignment comment" above
OpMsgBlock* messageBlock = reinterpret_cast<OpMsgBlock*>(eback());
// We now have at least the Message header block. So we know the message size.
std::size_t maxAvailableData = inMessageSize - messageReadSize();
if (underflowNeeds > maxAvailableData)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "underflow", "Should never need to read more bytes than in the message! Message Size: ", inMessageSize, "already read: ", (gp
return traits::eof();
}
// Try and read as much of the message as we can (whole message in one go if possible).
// But we may not need everything so don't block if it is not available.
// Note this read could be zero sized if we just read a compressed block.
if (missingData() != 0)
{
ThorsSocket::IOData result = getSocket().tryGetMessageData(egptr(), missingData());
newDataCount += result.dataSize;
extraData += result.dataSize;
// If we still don't have enough data to satisfy the underlying request
// Then we need to block and wait until we have it.
if (newDataCount < underflowNeeds && result.stillOpen)
{
// Must get at least 'underflowNeeds' bytes.
// So if not enough data was retrieved read blocking until we have data or there is an error
result = getSocket().getMessageData(egptr(), (underflowNeeds - newDataCount));
newDataCount += result.dataSize;
extraData += result.dataSize;
}
if (newDataCount < underflowNeeds)
{
ThorsLogCritical(
"ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer",
"underflow",
"Should never need to read more bytes than in the message! Message Size: ",
inMessageSize,
" requested bytes: ",
underflowNeeds);
return traits::eof();
}
}
setg(eback(), gptr(), egptr() + extraData);
if (missingData() == 0 && messageBlock->flags & OP_MsgFlag::checksumPresent)
{
// Calculate the expected checksum.
Checksum checksum;
checksum.append(std::string_view(eback(), inMessageSize - kSizeUInt32));
std::uint32_t checksumValue = checksum.checksum();
// Retrieve the checksum provided on the stream.
std::uint32_t checksumFromStream;
std::copy(egptr() - kSizeUInt32, egptr(), reinterpret_cast<char*>(&checksumFromStream));
// XOR the two values.
// Put that back on the stream.
// The next level up will then see a zero as success and anything else as failure.
std::uint32_t check = checksumValue ^ checksumFromStream;
std::copy(reinterpret_cast<char*>(&check), reinterpret_cast<char*>(&check) + kSizeUInt32, egptr() - kSizeUInt32);
if (check != 0)
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "underflow", "Checksum does not match expected checksum");
}
}
}
return gptr() == egptr() ? traits::eof() : traits::to_int_type(*gptr());
}
// Used by `xsgetn` to set the "underflowNeeds" value but
// put it back to the original value after the call to underflow().
struct UnderflowCountSetter
{
std::uint32_t& underflowNeeds;
UnderflowCountSetter(std::uint32_t& underflowNeeds, std::uint32_t value)
: underflowNeeds(underflowNeeds)
{
underflowNeeds = value;
}
~UnderflowCountSetter()
{
underflowNeeds = 1;
}
};
std::streamsize ConnectionBufferMongo::xsgetn(char_type* dest, std::streamsize count)
{
/*
* Reads count characters from the input sequence and stores them into a character array pointed to by dest.
*
* The characters are read as if by repeated calls to sbumpc().
*
* That is, if less than count characters are immediately available, the function calls uflow() to
* provide more until traits::eof() is returned. Classes derived from std::basic_streambuf are permitted
* to provide more efficient implementations of this function.
*/
// Check to see if the request is a valid length.
// Note: If messageLeft == 0 then we can go to the next message.
std::uint32_t messageLeft = inMessageSize - messageReadSize();
if (messageLeft != 0 && count > messageLeft)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "xsgetn", "Can not read more data than left in the message: requested: ", count, " data left: ", messageLeft);
return 0;
}
// See how much we have locally in the buffer.
std::streamsize currentBufferSize = availableData();
if (currentBufferSize < count)
{
UnderflowCountSetter set(underflowNeeds, (count - currentBufferSize));
if (underflow() == traits::eof())
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "xsgetn", "Underflow failed to retrieve enough data");
return 0;
}
}
currentBufferSize = egptr() - gptr();
std::uint32_t readable = std::min(count, currentBufferSize);
std::copy(gptr(), gptr() + readable, dest);
gbump(readable);
return readable;
}
ConnectionBufferMongo::int_type ConnectionBufferMongo::overflow(int_type ch)
{
if (ch == traits::eof()) {
return traits_type::to_int_type('a');
}
// Should not call overflow when message size is not zero.
// Writes fill up the buffer. Then sync() will send the data
// out to the stream and reset outMessageSize to zero.
if (outMessageSize != 0)
{
ThorsLogCritical(
"ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer",
"overflow",
"Unexpected call to overflow. The outMessageSize is set and we reservered space. outMessageSize: ",
outMessageSize, " Buffer: [pbase(): " , pbase(), " pptr(): " , pptr(), " epptr: ", epptr(), "]");
return traits::eof();
}
// There may be multiple calls to overflow until we have the size of the buffer stored.
pbase()[outMessagePlaced] = ch;
++outMessagePlaced;
if (outMessagePlaced == kSizeUInt32)
{
std::copy(pbase(), pbase() + kSizeUInt32, reinterpret_cast<char*>(&outMessageSize));
if (outMessageSize < kSizeSmallestPossibleMessage)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "overflow", "Minimum message size is 26 bytes.");
return traits::eof();
}
// OK. So we have the message size.
// Make sure we have enough space for the message and set up the buffer.
reserveOutputSize(outMessageSize);
setp(pbase(), pbase() + outMessageSize);
}
return ch;
}
std::streamsize ConnectionBufferMongo::xsputn(char_type const* source, std::streamsize count)
{
/*
* Writes count characters to the output sequence from the character array whose first element is pointed to by source.
*
* The characters are written as if by repeated calls to sputc().
* Writing stops when either count characters are written or a call to sputc() would have returned Traits::eof().
*
* If the put area becomes full (pptr() == epptr()), this function may call overflow(),
* or achieve the effect of calling overflow() by some other, unspecified, means.
*/
std::size_t sizeCount = 0;
while (outMessageSize == 0 && outMessagePlaced < kSizeUInt32)
{
if (overflow(source[sizeCount]) == traits::eof()) {
return 0;
}
++sizeCount;
}
std::streamsize spaceInBuffer = epptr() - pptr();
if (count > spaceInBuffer)
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "xsputn", "Failed to put data in buffer as it was not large enough. ExtraSpaceneeded: ", (count - spaceInBuffer));
return 0;
}
// Put everything in the buffer.
// We don't write until we have the complete message.
// This is because we may want to compress the buffer before sending (most Mongo servers support this).
std::copy(source + sizeCount, source + count, pptr() + sizeCount);
pbump(count);
return count;
}
int ConnectionBufferMongo::sync()
{
if (pptr() != epptr())
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::MessageBasedConnectionBuffer", "sync", "unexpected sync! The buffer is not yet full");
return -1;
}
OpMsgBlock* messageBlock = reinterpret_cast<OpMsgBlock*>(pbase());
if (messageBlock->flags & OP_MsgFlag::checksumPresent)
{
// The checksum pushed by the user is simply overwritten.
// There is no way for them to calculate the info so the expectation is that they write
// a 32 bit value and the end of the message. See Op_Msg
Checksum checksum;
checksum.append(std::string_view(pbase(), outMessageSize - kSizeUInt32));
std::uint32_t checksumValue = checksum.checksum();
std::copy(reinterpret_cast<char*>(&checksumValue), reinterpret_cast<char*>(&checksumValue) + kSizeUInt32, epptr() - kSizeUInt32);
}
if (useCompressionOnWrite)
{
std::string output;
std::uint32_t objectToCompressSize = (pptr() - pbase()) - kSizeOpMsgHeaderSize;
snappy::Compress(pbase() + kSizeOpMsgHeaderSize, objectToCompressSize, &output);
OpCompressedBlock compressedMessageHeader;
// Get a copy of the original header
std::copy(pbase(), pbase() + kSizeOpMsgHeaderSize, reinterpret_cast<char*>(&compressedMessageHeader));
// Now convert that to a compressed message header.
compressedMessageHeader.originalOpCode = compressedMessageHeader.opCode;
compressedMessageHeader.opCode = OpCode::OP_COMPRESSED;
compressedMessageHeader.uncompressedSize = compressedMessageHeader.messageSize - kSizeOpMsgHeaderSize;
compressedMessageHeader.compression = Compression::Snappy;
compressedMessageHeader.messageSize = kSizeOpCompressedBlock + output.size();
// output the compressed message.
writeToStream(reinterpret_cast<char const*>(&compressedMessageHeader), kSizeOpCompressedBlock);
writeToStream(&output[0], output.size());
}
else
{
// output the normal message
writeToStream(pbase(), epptr() - pbase());
}
outMessageSize = 0;
outMessagePlaced = 0;
setp(pbase(), pbase());
return 0;
}
test/ConnectionBufferMongoTest.cpp
#include "gtest/gtest.h"
#include "ConnectionBufferMongo.h"
#include "MongoUtil.h"
#include <unistd.h>
#include <iostream>
#include <fstream>
using ThorsAnvil::DB::Mongo::ConnectionBufferMongo;
using ThorsAnvil::DB::Mongo::OpMsgHeader;;
using ThorsAnvil::DB::Mongo::OP_MsgFlag;
using ThorsAnvil::DB::Mongo::OpCode;
using ThorsAnvil::ThorsSocket::Open;
using ThorsAnvil::ThorsSocket::Mode;
TEST(ConnectionBufferMongoTest, Build)
{
char data[10000] = {0};
getcwd(data, 10000);
ConnectionBufferMongo buffer({"test/data/emptyMessage", Open::Append});
EXPECT_NE(0, buffer.getSocket().socketId(Mode::Read));
}
TEST(ConnectionBufferMongoTest, ReadHeader)
{
OpMsgHeader data{0, 0, 0, OpCode::OP_NOOP};
ConnectionBufferMongo buffer({"test/data/emptyMessage", ThorsAnvil::ThorsSocket::Open::Append});
auto val = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(26, data.messageSize);
EXPECT_EQ(15, data.messageId);
EXPECT_EQ(32, data.messageResponseId);
EXPECT_EQ(OpCode::OP_MSG, data.opCode);
}
TEST(ConnectionBufferMongoTestTest, BadMessageHeader)
{
OpMsgHeader data{0, 0, 0, OpCode::OP_NOOP};
ConnectionBufferMongo buffer({"test/data/badMessageSize", ThorsAnvil::ThorsSocket::Open::Append});
auto val = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(0, val);
}
TEST(ConnectionBufferMongoTest, ReadMoreMessagesThanIsAvailable)
{
char data[50];
ConnectionBufferMongo buffer({"test/data/emptyMessages", ThorsAnvil::ThorsSocket::Open::Append});
auto val = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(0, val);
}
TEST(ConnectionBufferMongoTest, ReadTwoMessagesSoPubSeekIsIncremented)
{
char dataIgnore[26];
ConnectionBufferMongo buffer({"test/data/twoEmptyMessages", ThorsAnvil::ThorsSocket::Open::Append});
auto val1 = buffer.sgetn(reinterpret_cast<char*>(&dataIgnore), sizeof(dataIgnore));
EXPECT_EQ(26, val1);
EXPECT_EQ(26, buffer.pubseekoff(0, std::ios::cur,std::ios::in));
OpMsgHeader data{0, 0, 0, OpCode::OP_NOOP};
auto val2 = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, val2);
EXPECT_EQ(26, data.messageSize);
EXPECT_EQ(15, data.messageId);
EXPECT_EQ(32, data.messageResponseId);
EXPECT_EQ(OpCode::OP_MSG, data.opCode);
EXPECT_EQ(42, buffer.pubseekoff(0, std::ios::cur,std::ios::in));
}
TEST(ConnectionBufferMongoTest, ReadMoreThanBufferHasButWithinRange)
{
OpMsgHeader data{0, 0, 0, OpCode::OP_NOOP};
ConnectionBufferMongo buffer({"test/data/messagesHeaderAndFlag", ThorsAnvil::ThorsSocket::Open::Append});
auto val = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, val);
// There are 4 bytes in the buffer.
// There should be 10 bytes in the object but the file will fail to read more than 4.
char dataIngore[10];
auto val2 = buffer.sgetn(reinterpret_cast<char*>(&dataIngore), sizeof(dataIngore));
EXPECT_EQ(4, val2);
}
TEST(ConnectionBufferMongoTest, WriteMessage)
{
OpMsgHeader data{26, 15, 32, OpCode::OP_MSG};
ConnectionBufferMongo buffer({"/tmp/WriteMessage", ThorsAnvil::ThorsSocket::Open::Truncate});
auto val = buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, val);
}
TEST(ConnectionBufferMongoTest, WriteMessageFlushBeforeMessageComplete)
{
OpMsgHeader data{26, 15, 32, OpCode::OP_MSG};
ConnectionBufferMongo buffer({"/tmp/WriteMessage", ThorsAnvil::ThorsSocket::Open::Truncate});
auto val = buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, val);
auto s = buffer.pubsync();
EXPECT_EQ(-1, s);
}
TEST(ConnectionBufferMongoTest, WriteMessageFlush)
{
OpMsgHeader data{26, 15, 32, OpCode::OP_MSG};
ConnectionBufferMongo buffer({"/tmp/WriteMessage", ThorsAnvil::ThorsSocket::Open::Truncate});
auto val = buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, val);
OP_MsgFlag flags = OP_MsgFlag::empty;
val = buffer.sputn(reinterpret_cast<char*>(&flags), sizeof(flags));
EXPECT_EQ(4, val);
std::uint8_t kind = 0;
val = buffer.sputn(reinterpret_cast<char*>(&kind), sizeof(kind));
EXPECT_EQ(1, val);
std::int32_t bsonSize = 5;
val = buffer.sputn(reinterpret_cast<char*>(&bsonSize), sizeof(bsonSize));
EXPECT_EQ(4, val);
std::uint8_t bsonTerm = 0;
val = buffer.sputn(reinterpret_cast<char*>(&bsonTerm), sizeof(bsonTerm));
EXPECT_EQ(1, val);
auto s = buffer.pubsync();
EXPECT_EQ(0, s);
}
struct LocalOpMsg: public OpMsgHeader
{
char buffer[1000];
};
TEST(ConnectionBufferMongoTest, WriteMessageMoreThanBuffer)
{
LocalOpMsg data{26, 15, 32, OpCode::OP_MSG};
ConnectionBufferMongo buffer({"/tmp/WriteMessage", ThorsAnvil::ThorsSocket::Open::Truncate});
auto val = buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(0, val);
}
TEST(ConnectionBufferMongoTest, CheckThatMoveWorks)
{
LocalOpMsg data{26, 15, 32, OpCode::OP_MSG};
ConnectionBufferMongo buffer({"/tmp/WriteMessage", ThorsAnvil::ThorsSocket::Open::Truncate});
auto val = buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
auto tel = buffer.pubseekoff(0, std::ios::cur,std::ios::in);
ConnectionBufferMongo buffer2(std::move(buffer));
EXPECT_EQ(tel, buffer2.pubseekoff(0, std::ios::cur,std::ios::in));
}
/*
Use this test to build data for the test: ReadCompressData
*/
TEST(ConnectionBufferMongoTest, WriteCompressData)
{
{
ConnectionBufferMongo buffer({"/tmp/compressedData", ThorsAnvil::ThorsSocket::Open::Truncate});
buffer.setCompressionOnWrite();
OpMsgHeader data{26, 15, 32, OpCode::OP_MSG};
buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
OP_MsgFlag flags = OP_MsgFlag::empty;
buffer.sputn(reinterpret_cast<char*>(&flags), sizeof(flags));
std::uint8_t kind = 0;
buffer.sputn(reinterpret_cast<char*>(&kind), sizeof(kind));
std::int32_t bsonSize = 5;
buffer.sputn(reinterpret_cast<char*>(&bsonSize), sizeof(bsonSize));
std::uint8_t bsonTerm = 0;
buffer.sputn(reinterpret_cast<char*>(&bsonTerm), sizeof(bsonTerm));
buffer.pubsync();
}
std::ifstream dataFile("/tmp/compressedData", std::ios::binary);
std::vector<char> data{std::istreambuf_iterator<char>{dataFile}, std::istreambuf_iterator<char>{}};
std::vector<char> expected{
'\x25', '\x00', '\x00', '\x00', // Message Size
'\x0f', '\x00', '\x00', '\x00', // Message Id
'\x20', '\x00', '\x00', '\x00', // Response Id
'\xdc', '\x07', '\x00', '\x00', // Op Code (OP_COMPRESSED)
'\xdd', '\x07', '\x00', '\x00', // Original OpCode (OP_MSG)
'\x0a', '\x00', '\x00', '\x00', // Original Message Size
'\x01', // Compression Type (Snappy)
'\x0a', '\x24', '\x00', '\x00', '\x00', '\x00', '\x00', '\x05', '\x00', '\x00', '\x00', '\x00'};
EXPECT_EQ(expected, data);
}
TEST(ConnectionBufferMongoTest, ReadCompressData)
{
ConnectionBufferMongo buffer({"test/data/compressedData", ThorsAnvil::ThorsSocket::Open::Append});
OpMsgHeader data{0, 0, 0, OpCode::OP_NOOP};
auto r = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, r);
EXPECT_EQ(26, data.messageSize);
EXPECT_EQ(15, data.messageId);
EXPECT_EQ(32, data.messageResponseId);
EXPECT_EQ(OpCode::OP_MSG, data.opCode);
OP_MsgFlag flags = static_cast<OP_MsgFlag>(255);
r = buffer.sgetn(reinterpret_cast<char*>(&flags), sizeof(flags));
EXPECT_EQ(4, r);
EXPECT_EQ(OP_MsgFlag::empty, flags);
std::uint8_t kind = 12;
r = buffer.sgetn(reinterpret_cast<char*>(&kind), sizeof(kind));
EXPECT_EQ(1, r);
EXPECT_EQ(0, kind);
std::int32_t bsonSize = 255;
r = buffer.sgetn(reinterpret_cast<char*>(&bsonSize), sizeof(bsonSize));
EXPECT_EQ(4, r);
EXPECT_EQ(5, bsonSize);
std::uint8_t bsonTerm = 1;
r = buffer.sgetn(reinterpret_cast<char*>(&bsonTerm), sizeof(bsonTerm));
EXPECT_EQ(1, r);
EXPECT_EQ(0, bsonTerm);
}
TEST(ConnectionBufferMongoTest, ReadDataWithChecksum)
{
ConnectionBufferMongo buffer({"test/data/emptyMessageWithChecksum", ThorsAnvil::ThorsSocket::Open::Append});
OpMsgHeader data{0, 0, 0, OpCode::OP_NOOP};
auto r = buffer.sgetn(reinterpret_cast<char*>(&data), sizeof(data));
EXPECT_EQ(16, r);
EXPECT_EQ(30, data.messageSize);
EXPECT_EQ(15, data.messageId);
EXPECT_EQ(32, data.messageResponseId);
EXPECT_EQ(OpCode::OP_MSG, data.opCode);
OP_MsgFlag flags = static_cast<OP_MsgFlag>(255);
r = buffer.sgetn(reinterpret_cast<char*>(&flags), sizeof(flags));
EXPECT_EQ(4, r);
EXPECT_EQ(OP_MsgFlag::checksumPresent, flags);
std::uint8_t kind = 12;
r = buffer.sgetn(reinterpret_cast<char*>(&kind), sizeof(kind));
EXPECT_EQ(1, r);
EXPECT_EQ(0, kind);
std::int32_t bsonSize = 255;
r = buffer.sgetn(reinterpret_cast<char*>(&bsonSize), sizeof(bsonSize));
EXPECT_EQ(4, r);
EXPECT_EQ(5, bsonSize);
std::uint8_t bsonTerm = 1;
r = buffer.sgetn(reinterpret_cast<char*>(&bsonTerm), sizeof(bsonTerm));
EXPECT_EQ(1, r);
EXPECT_EQ(0, bsonTerm);
std::uint32_t checksum = 1;
r = buffer.sgetn(reinterpret_cast<char*>(&checksum), sizeof(checksum));
EXPECT_EQ(4, r);
EXPECT_EQ(0, checksum);
}
/*
Use this test to build data for the test: ReadDataWithChecksum
*/
TEST(ConnectionBufferMongoTest, WriteDataWithChecksum)
{
{
ConnectionBufferMongo buffer({"/tmp/checksumData", ThorsAnvil::ThorsSocket::Open::Truncate});
OpMsgHeader data{30, 15, 32, OpCode::OP_MSG};
buffer.sputn(reinterpret_cast<char*>(&data), sizeof(data));
OP_MsgFlag flags = OP_MsgFlag::checksumPresent;
buffer.sputn(reinterpret_cast<char*>(&flags), sizeof(flags));
std::uint8_t kind = 0;
buffer.sputn(reinterpret_cast<char*>(&kind), sizeof(kind));
std::int32_t bsonSize = 5;
buffer.sputn(reinterpret_cast<char*>(&bsonSize), sizeof(bsonSize));
std::uint8_t bsonTerm = 0;
buffer.sputn(reinterpret_cast<char*>(&bsonTerm), sizeof(bsonTerm));
std::uint32_t checksum = 0;
buffer.sputn(reinterpret_cast<char*>(&checksum), sizeof(checksum));
buffer.pubsync();
}
std::ifstream dataFile("/tmp/checksumData", std::ios::binary);
std::vector<char> data{std::istreambuf_iterator<char>{dataFile}, std::istreambuf_iterator<char>{}};
std::vector<char> expected{
'\x1E', '\x00', '\x00', '\x00', // Message Size
'\x0f', '\x00', '\x00', '\x00', // Message Id
'\x20', '\x00', '\x00', '\x00', // Response Id
'\xdd', '\x07', '\x00', '\x00', // Op Code (OP_MSG)
'\x01', '\x00', '\x00', '\x00', // Flag => Checksum bit set
'\x00', // Kind 0
'\x05', '\x00', '\x00', '\x00', // BSON object size.
'\x00', // BSON object terminator
'\x44', '\x11', '\x37', '\xb8' // Checksum
};
// Checksum generated on this page: http://www.sunshine2k.de/coding/javascript/crc/crc_js.html
// Using CRC32_C (Default parameters)
// Places the following byetes in the CRC Input Box (Select Bytes)
// 0x1E 0x00 0x00 0x00 0x0f 0x00 0x00 0x00 0x20 0x00 0x00 0x00 0xdd 0x07 0x00 0x00 0x01 0x00 0x00 0x00 0x00 0x05 0x00 0x00 0x00 0x00
// Click "Calculate CRC": Result => 0xB8371144
EXPECT_EQ(expected, data);
}
