Refactored part of the original code:
This review is looking only at the "Op_MSG()" functions. These functions are used to read/write standard C++ objects to Mongo. The C++ objects are converted into BSON by ThorsSerializer (thus requires no coding).
Note 1: These functions DO NOT assume that the underlying stream is using "ConnectionBufferMongo" (see previous review). But they do assume that stream only has OP_MSG objects on the stream (and not any compressed objects).
Note 2: It will not handle the the checksum itself it does rely on the underlying buffer for that. But can be set to ignore the checksum if it does not know how to handle it.
Usage would be:
template<typename T>
void sendMessage(std::ostream const& stream, std::uint32_t messageId, T const& messageToSendToMongo)
{
stream << Op_Msg(messageToSendToMongo, messageId);
}
template<typename T>
void getMessage(std::ostream& stream, std::uint32_t responseToMessage, T& messageFromMongo)
{
stream >> Op_Msg(messageFromMongo, responseToMessage);
}
Op_Msg.h
#ifndef THORSANVIL_DB_MONGO_OP_MSG_H
#define THORSANVIL_DB_MONGO_OP_MSG_H
#include "ThorsMongoConfig.h"
#include "MongoUtil.h"
#include <bit>
static_assert(std::endian::little == std::endian::native,
"Don't want to support big endian unless I can do some good testing. "
"Don't have a system for that so currently not suppoorted, "
"but should be a relatively easy change");
#include <iostream>
#include "ThorSerialize/SerUtil.h"
#include "ThorSerialize/BsonThor.h"
namespace ThorsAnvil::DB::Mongo
{
using MessageConfig = ThorsAnvil::Serialize::ParserInterface::ParserConfig;
template<typename Section>
class Op_MsgOut
{
private:
static MessageId getNextMessageId()
{
static MessageId nextMessageId = 0;
return nextMessageId++;
}
std::uint32_t calcSize() const
{
return kSizeOpMsgBlock
+ ThorsAnvil::Serialize::bsonGetPrintSize(section)
// Op_Msg Optional Checksum
+ (hasCheckSum() ? sizeof(std::int32_t) : 0);
}
bool hasCheckSum() const {return (flags & OP_MsgFlag::checksumPresent) != 0;}
OP_MsgFlag getFlags() const {return flags;}
Section const& section;
MessageId& messageId;
OP_MsgFlag flags;
bool handleChecksum;
public:
Op_MsgOut(Section const& section, MessageId& messageId, OP_MsgFlag flags, bool handleChecksum)
: section(section)
, messageId(messageId)
, flags(flags)
, handleChecksum(handleChecksum)
{}
friend std::ostream& operator<<(std::ostream& stream, Op_MsgOut const& message)
{
OpMsgBlock wireMessage;
wireMessage.messageSize = message.calcSize();
wireMessage.messageId = message.getNextMessageId();
wireMessage.messageResponseId = 0;
wireMessage.opCode = OpCode::OP_MSG;
wireMessage.flags = message.getFlags();
wireMessage.kind = 0;
if (!message.handleChecksum)
{
// If we can't handle checksum then don't allow the flag to be set.
wireMessage.flags &= ~OP_MsgFlag::checksumPresent;
}
stream.write(reinterpret_cast<char*>(&wireMessage), kSizeOpMsgBlock)
<< ThorsAnvil::Serialize::bsonExporter(message.section);
if (message.hasCheckSum())
{
// That actual checksum creation is done in ConnectionBufferMongo::sync()
// The sync happens when we flush the stream below.
std::uint32_t checksum = 0;
stream.write(reinterpret_cast<char*>(&checksum), sizeof(checksum));
}
if (stream)
{
message.messageId = wireMessage.messageId;
}
else
{
ThorsLogCritical("ThorsAnvil::DB::Mongo::Op_MsgOut", "operator<<", "Serialization of message failed.");
}
return stream << std::flush;
}
};
template<typename Section>
class Op_MsgIn
{
private:
Section& section;
MessageConfig const& config;
MessageId messageId;
bool handleChecksum;
public:
Op_MsgIn(Section& section, MessageConfig const& config, MessageId messageId, bool handleChecksum)
: section(section)
, config(config)
, messageId(messageId)
, handleChecksum(handleChecksum)
{}
friend std::istream& operator>>(std::istream& stream, Op_MsgIn const& message)
{
OpMsgBlock wireMessage;
std::streampos start = stream.tellg();
stream.read(reinterpret_cast<char*>(&wireMessage), kSizeOpMsgBlock)
>> ThorsAnvil::Serialize::bsonImporter(message.section, message.config);
if (stream)
{
if (wireMessage.flags & OP_MsgFlag::checksumPresent)
{
// That actual checksum validation is done in ConnectionBufferMongo::underflow()
std::uint32_t checksum;
stream.read(reinterpret_cast<char*>(&checksum), sizeof(checksum));
// If validation worked correctly. Then it XOR the calculated value on the stream.
// Thus if everything worked it will result in this checksum being zero.
// If we don't want to handle checksum then simply ignore it.
if (message.handleChecksum && checksum != 0)
{
stream.setstate(std::ios::failbit);
ThorsLogCritical("ThorsAnvil::DB::Mongo::Op_MsgIn", "operator>>", "Deserialization checksum validation failed");
}
}
std::streampos end = stream.tellg();
std::size_t size = end - start;
if (size != wireMessage.messageSize)
{
stream.setstate(std::ios::failbit);
ThorsLogCritical("ThorsAnvil::DB::Mongo::Op_MsgIn", "operator>>", "Deserialization of message failed. Invalid Size: Bytes read: ", size, " Wire Message Size: ", wireMes
}
if (wireMessage.opCode != OpCode::OP_MSG)
{
stream.setstate(std::ios::failbit);
ThorsLogCritical("ThorsAnvil::DB::Mongo::Op_MsgIn", "operator>>", "Unexpected Message Type. Expected: OP_MSG(", static_cast<int>(OpCode::OP_MSG), ") Got(", static_cast<
}
if (wireMessage.messageResponseId != message.messageId)
{
stream.setstate(std::ios::failbit);
ThorsLogCritical("ThorsAnvil::DB::Mongo::Op_MsgIn", "operator>>", "This message is not the expected reply: Expected: ", message.messageId, " Got: ", wireMessage.message
}
}
return stream;
}
};
template<typename Section>
Op_MsgOut<Section> Op_Msg(Section const& section, MessageId& messageId, OP_MsgFlag flags = OP_MsgFlag::empty, bool handleChecksum = true)
{
return Op_MsgOut<Section>(section, messageId, flags, handleChecksum);
}
template<typename Section>
Op_MsgIn<Section> Op_Msg(Section& section, MessageId messageId, MessageConfig const& config = MessageConfig{}, bool handleChecksum = true)
{
return Op_MsgIn<Section>(section, config, messageId, handleChecksum);
}
}
#endif
test/Op_MsgTest.cpp
#include "gtest/gtest.h"
#include "Op_Msg.h"
#include "ThorSerialize/Traits.h"
#include <sstream>
#include <algorithm>
#include <iomanip>
class Empty {};
ThorsAnvil_MakeTrait(Empty);
using ThorsAnvil::DB::Mongo::Op_Msg;
using ThorsAnvil::DB::Mongo::OP_MsgFlag;
using ThorsAnvil::DB::Mongo::MessageId;
using ThorsAnvil::DB::Mongo::MessageConfig;
TEST(OP_MsgTest, OutputEmptyMessage)
{
std::stringstream stream;
MessageId id;
stream << Op_Msg(Empty{}, id);
EXPECT_TRUE(static_cast<bool>(stream));
std::string result = stream.str();
EXPECT_EQ(26, result.size());
// Assumages native int is little endian
std::int32_t messageSize;
std::copy(std::begin(result) + 00, std::begin(result) + 04, reinterpret_cast<char*>(&messageSize));
// Should be Message Header (4 * std::int32_t) + Flags (std::int32_t) + Kind (char) + Document
// Empty document is int32_t + char (0)
// 16 + 4 + 1 + 4 + 1 => 26
EXPECT_EQ(26, messageSize);
EXPECT_EQ(26, result.size());
std::int32_t messageOp;
std::copy(std::begin(result) + 12, std::begin(result) + 16, reinterpret_cast<char*>(&messageOp));
EXPECT_EQ(2013, messageOp); // OP_MSG
OP_MsgFlag messageFlag;
std::copy(std::begin(result) + 16, std::begin(result) + 20, reinterpret_cast<char*>(&messageFlag));
EXPECT_EQ(OP_MsgFlag::empty, messageFlag);
char kind = result[20];
EXPECT_EQ(0, kind);
}
TEST(OP_MsgTest, OutputMessageExplicitEmptyFlag)
{
std::stringstream stream;
MessageId id;
stream << Op_Msg(Empty{}, id, OP_MsgFlag::empty);
EXPECT_TRUE(static_cast<bool>(stream));
std::string result = stream.str();
EXPECT_EQ(26, result.size());
OP_MsgFlag messageFlag;
std::copy(std::begin(result) + 16, std::begin(result) + 20, reinterpret_cast<char*>(&messageFlag));
EXPECT_EQ(OP_MsgFlag::empty, messageFlag);
}
TEST(OP_MsgTest, OutputMessageExplicitChecksumPresentFlag)
{
std::stringstream stream;
MessageId id;
stream << Op_Msg(Empty{}, id, OP_MsgFlag::checksumPresent);
EXPECT_TRUE(static_cast<bool>(stream));
std::string result = stream.str();
EXPECT_EQ(30, result.size()); // Note 4 more bytes
OP_MsgFlag messageFlag;
std::copy(std::begin(result) + 16, std::begin(result) + 20, reinterpret_cast<char*>(&messageFlag));
EXPECT_EQ(OP_MsgFlag::checksumPresent, messageFlag);
}
TEST(OP_MsgTest, OutputMessageExplicitMoreToComeFlag)
{
std::stringstream stream;
MessageId id;
stream << Op_Msg(Empty{}, id, OP_MsgFlag::moreToCome);
EXPECT_TRUE(static_cast<bool>(stream));
std::string result = stream.str();
EXPECT_EQ(26, result.size());
OP_MsgFlag messageFlag;
std::copy(std::begin(result) + 16, std::begin(result) + 20, reinterpret_cast<char*>(&messageFlag));
EXPECT_EQ(OP_MsgFlag::moreToCome, messageFlag);
}
TEST(OP_MsgTest, OutputMessageExplicitExhaustAllowedFlag)
{
std::stringstream stream;
MessageId id;
stream << Op_Msg(Empty{}, id, OP_MsgFlag::exhaustAllowed);
EXPECT_TRUE(static_cast<bool>(stream));
std::string result = stream.str();
EXPECT_EQ(26, result.size());
OP_MsgFlag messageFlag;
std::copy(std::begin(result) + 16, std::begin(result) + 20, reinterpret_cast<char*>(&messageFlag));
EXPECT_EQ(OP_MsgFlag::exhaustAllowed, messageFlag);
}
TEST(OP_MsgTest, OutputMessageAndIgnoreChecksum)
{
std::stringstream stream;
MessageId id;
stream << Op_Msg(Empty{}, id, OP_MsgFlag::exhaustAllowed, false);
EXPECT_TRUE(static_cast<bool>(stream));
std::string result = stream.str();
EXPECT_EQ(26, result.size());
OP_MsgFlag messageFlag;
std::copy(std::begin(result) + 16, std::begin(result) + 20, reinterpret_cast<char*>(&messageFlag));
EXPECT_EQ(OP_MsgFlag::exhaustAllowed, messageFlag);
}
//===
TEST(OP_MsgTest, InputEmptyMessage)
{
using namespace std::string_literals;
std::istringstream stream( "\x1a\x00\x00\x00"s // Size
"\x0e\x00\x00\x00"s // MessageId
"\x20\x00\x00\x00"s // ResponseId
"\xdd\x07\x00\x00"s // OP_MSG
"\x00\x00\x00\x00"s // Flags: empty
"\x00"s // Kind 0
"\x05\x00\x00\x00"s // BSON size
"\x00"s // BSON terminator
);
Empty data;
stream >> Op_Msg(data, 0x20);
EXPECT_TRUE(static_cast<bool>(stream));
}
TEST(OP_MsgTest, InputMessageChecksumPresent)
{
using namespace std::string_literals;
std::istringstream stream( "\x1e\x00\x00\x00"s // Size
"\x0e\x00\x00\x00"s // MessageId
"\x20\x00\x00\x00"s // ResponseId
"\xdd\x07\x00\x00"s // OP_MSG
"\x01\x00\x00\x00"s // Flags: Checksum
"\x00"s // Kind 0
"\x05\x00\x00\x00"s // BSON size
"\x00"s // BSON terminator
"\x44\x11\x37\xb8"s // Checksum
);
Empty data;
stream >> Op_Msg(data, 0x20);
EXPECT_FALSE(static_cast<bool>(stream)); // Checksum failure.
// As the stringstream does not use ConnectionBufferMongo
stream.clear();
EXPECT_TRUE(static_cast<bool>(stream));
EXPECT_EQ(30, stream.tellg()); // Even though checksum failed.
// We should have read the whole message.
}
TEST(OP_MsgTest, InputMessageChecksumPresentIgnoreChecksum)
{
using namespace std::string_literals;
std::istringstream stream( "\x1e\x00\x00\x00"s // Size
"\x0e\x00\x00\x00"s // MessageId
"\x20\x00\x00\x00"s // ResponseId
"\xdd\x07\x00\x00"s // OP_MSG
"\x01\x00\x00\x00"s // Flags: Checksum
"\x00"s // Kind 0
"\x05\x00\x00\x00"s // BSON size
"\x00"s // BSON terminator
"\x44\x11\x37\xb8"s // Checksum
);
Empty data;
stream >> Op_Msg(data, 0x20, MessageConfig{}, false);
EXPECT_TRUE(static_cast<bool>(stream));
}
TEST(OP_MsgTest, InputMessageMoreToCome)
{
using namespace std::string_literals;
std::istringstream stream( "\x1a\x00\x00\x00"s // Size
"\x0e\x00\x00\x00"s // MessageId
"\x20\x00\x00\x00"s // ResponseId
"\xdd\x07\x00\x00"s // OP_MSG
"\x02\x00\x00\x00"s // Flags: moreToCome
"\x00"s // Kind 0
"\x05\x00\x00\x00"s // BSON size
"\x00"s // BSON terminator
);
Empty data;
stream >> Op_Msg(data, 0x20);
EXPECT_TRUE(static_cast<bool>(stream));
}
TEST(OP_MsgTest, InputMessageExhaustAllowed)
{
using namespace std::string_literals;
std::istringstream stream( "\x1a\x00\x00\x00"s // Size
"\x0e\x00\x00\x00"s // MessageId
"\x20\x00\x00\x00"s // ResponseId
"\xdd\x07\x00\x00"s // OP_MSG
"\x00\x00\x01\x00"s // Flags: exhaustAllowed
"\x00"s // Kind 0
"\x05\x00\x00\x00"s // BSON size
"\x00"s // BSON terminator
);
Empty data;
stream >> Op_Msg(data, 0x20);
EXPECT_TRUE(static_cast<bool>(stream));
}
TEST(OP_MsgTest, InputMessageInvalidMessageId)
{
using namespace std::string_literals;
std::istringstream stream( "\x1a\x00\x00\x00"s // Size
"\x0e\x00\x00\x00"s // MessageId
"\x20\x00\x00\x00"s // ResponseId
"\xdd\x07\x00\x00"s // OP_MSG
"\x00\x00\x01\x00"s // Flags: exhaustAllowed
"\x00"s // Kind 0
"\x05\x00\x00\x00"s // BSON size
"\x00"s // BSON terminator
);
Empty data;
stream >> Op_Msg(data, 0x21);
EXPECT_FALSE(static_cast<bool>(stream));
}