2
\$\begingroup\$

Refactored part of the original code:

c++ Mongo Interface

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));
  }
\$\endgroup\$

1 Answer 1

3
\$\begingroup\$

Usage

I'm not sure if you really meant there to be two functions sendMessage() and getMessage(). If not, then next time don't write suggested usage like this, just show us the example main() or test cases. However, if you do intend there to be these two functions, there are some issues with them:

  • They seem superfluous. Using sendMessage() is more typing than having the caller just write stream << Op_Msg(…) manually.
  • getMessage() should return the message instead of having an out parameter.
  • They always call Op_Msg() with just two parameters, the other parameters will always have their default values.

Naming things

The way you name things is inconsistent. Sometimes you use Msg, other times Message. I see Op_, OP_ and Op used as prefixes. It could be some of these consistencies are inherited from MongoDB's own code base or documentation, but it's best not to repeat the mistakes of others.

About Section

Looking at the MongoDB documentation, it seems that a single OP_MSG can contain multiple sections. Each section also contains a size and an identifier string. However, from your code it looks like you only support one section, and then only a "type 0" section.

Also, while the actual data you want to send/receive is in a section of an OP_MSG, I don't think you should name the (template) parameters section. Rather use T and data:

template<typename T>
class Op_MsgOut
{
    …
    T const& data;
    …
    Op_MsgOut(T const& data, …): data(data), … {}
    …
};

Be careful when storing references

One issue with Op_MsgIn and Op_MsgOut is that they store a reference to the data to be received or transmitted. What if that data goes out of scope? That's easier than you thing. Consider writing:

auto msg = Op_Msg(3.1415, 42);
stream << msg;

This deduces Section = double, and a reference to the temporary value 3.1415 is stored inside msg. However, that temporary value has gone out of scope when it tries to call operator<<.

You can try to prevent this case from happening by only letting operator<< take r-value references:

friend std::ostream& operator<<(std::ostream& stream, Op_MsgOut const&& message) {
    …
}

Although that doesn't prevent all possible misuses. Some more safety is gained by deleting the copy and move constructors and assignment operators of Op_MsgIn and Op_MsgOut.

Much safer (but perhaps less efficient) would be to not store a reference, but rather a copy of the data (either in the original form or BSON-encoded).

Minor issues

  • Avoid doing multiple things on one line, especially if you are mixing different styles. I had to read stream.write(…) << bsonExporter(…) three times before my mind parsed it correctly.
  • Don't left-pad numbers with zeroes, like in 00 and 04, as these will unfortunately be interpreted as octal numbers in C and C++. You'd have noticed the compile errors or test failures if you had written 08 or had added even more zero padding.
  • Don't flush streams unnecessarily. The caller can do it if so desired.
  • Your operator<<() will log a critical message on stream errors, but operator>>() does not.
\$\endgroup\$
2

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.