spark/src/lib/jetfix/codec/FIXEncodingStream.h

#pragma once

#include "lib/jetfix/codec/FIXBuffer.h"
#include "lib/jetfix/codec/FIXFragment.h"
#include "lib/jetfix/codec/FieldDefinitions.h"
#include "lib/jetfix/codec/Fields.h"
#include "lib/jetfix/codec/MessageTypes.h"
#include <memory>

#include "lib/config/protos/jetfix.pb.h"

namespace bts {
namespace jetfix {

// Helper to batch the encoding of multiple FIX messages all sharing
// the same FIX headers.
class FIXEncodingStream {
 public:
   FIXEncodingStream(config::JetFIX::Connection const &cfg,
                     uint32_t initial_capacity = 1024 * 10)
       : _msg_header_template(), _header_template_size(),
         _header_template_checksum(), _buffer(initial_capacity), _message_map(),
         _current_msg(_buffer), _state(State::OPEN), _seq_num_bm(),
         _last_seq_num_bm(), _sending_time_bm() {
      _message_map.reserve(10 * 1024);

      FIXBuffer header_buffer(&_msg_header_template[0],
                              _msg_header_template.size());
      FIXFragment header(header_buffer);

      // We cannot include body lenght and msg type in the header
      // because message type is not fixed lenght, and therefore its
      // not possible to have bookmarks to fields after it, ahead of
      // knowing the message type value. The rest of the header is
      // fixed lenght so we form it here.
      _seq_num_bm = header.add<fix::MsgSeqNum>(0, 9);
      _last_seq_num_bm = header.add<fix::LastSeqNumProcessed>(0, 9);
      _sending_time_bm = header.add<SendingTimeStr>(
          "20150427-19:05:18.923", sizeof("20150427-19:05:18.923") - 1);

      header.add<fix::SenderCompID>(cfg.sender_compid());
      header.add<fix::TargetCompId>(cfg.target_compid());
      header.add<fix::TargetSubID>(cfg.target_subid());
      _header_template_size = header.size();
      _header_template_checksum = header.checksum();
   }

   inline FIXFragment &new_msg(MessageTypes type) {
      bts_assert(_state == State::OPEN);
      tidy_last_msg();

      auto msg_offset = _buffer.size();

      _current_msg = FIXFragment(_buffer);
      _begin_string_bm = _current_msg.add<fix::BeginString>(std::string_view {"FIX.4.2"});
      _body_length_bm = _current_msg.add<fix::BodyLength>(0, 6);

      char mt_encoding[4];
      auto start = encode_message_type(type, mt_encoding);
      _msg_type_bm = _current_msg.add<fix::MsgType>(
          mt_encoding + start, sizeof(mt_encoding) - start);

      auto header_offset = _buffer.size();
      _message_map.emplace_back(msg_offset, header_offset, 0u, 0u, 0u,
                                FIXFragment::Bookmark<IntField<10u>>());
      _current_msg.append(&_msg_header_template[0], _header_template_size,
                          _header_template_checksum);
      return _current_msg;
   }

   inline void close() {
      bts_assert(_state == State::OPEN);
      tidy_last_msg();
      _state = State::CLOSED;
   }

   inline void reopen_empty() {
      bts_assert(_state == State::CLOSED);
      bts_assert(_message_map.empty());
      _state = State::OPEN;
   }

   inline uint32_t size() const { return _buffer.size(); }

   inline const char *data() const { return _buffer.data(); }

   struct MessageDesc {
      MessageDesc(uint32_t offset, uint32_t header_offset, uint32_t size,
                  uint32_t checksum, uint32_t seq_num,
                  FIXFragment::Bookmark<fix::Checksum> checksum_bm)
          : offset(offset), header_offset(header_offset), size(size),
            checksum(checksum), seq_num(seq_num), checksum_bm(checksum_bm) {}

      uint32_t offset;
      uint32_t header_offset;
      uint32_t size;
      uint32_t checksum;
      uint32_t seq_num;
      FIXFragment::Bookmark<fix::Checksum> checksum_bm;
   };

   std::string to_str() const {
      std::string result;

      if (size()) {
         // return std::string(data(),size() is simpler but valgrind
         // reports a warning in that call
         result.resize(size());
         std::copy(data(), data() + size(), result.begin());
      }
      return result;
   }

   std::string to_human_str() const {
      auto s = to_str();
      std::replace(s.begin(), s.end(), '\001', '|');
      return s;
   }

   inline uint32_t stamp(uint32_t seq_num, uint32_t last_processed,
                         std::string const &time_str) {
      bts_assert(_state == State::CLOSED);
      _state = State::STAMPED;

      for (auto &m : _message_map) {
         ++seq_num;
         m.seq_num = seq_num;

         FIXFragment frag(_buffer, m.checksum);
         frag.overwrite(_seq_num_bm.shift(m.header_offset), m.seq_num, 9);
         frag.overwrite(_last_seq_num_bm.shift(m.header_offset), last_processed,
                        9);
         frag.overwrite(_sending_time_bm.shift(m.header_offset), time_str);
         frag.overwrite_checksum(m.checksum_bm, frag.checksum() % 256);
      }
      return seq_num;
   }

   inline void reset() {
      _buffer.reset();
      _message_map.clear();
      _state = State::OPEN;
   }

   inline bool is_closed() const { return _state == State::CLOSED; }

   size_t n_msgs() const { return _message_map.size(); }

   std::vector<MessageDesc> const &msg_map() const { return _message_map; }

 private:
   std::array<char, 256> _msg_header_template;
   uint32_t _header_template_size;
   uint32_t _header_template_checksum;
   FIXBuffer _buffer;
   std::vector<MessageDesc> _message_map;
   FIXFragment _current_msg;
   enum class State { OPEN, CLOSED, STAMPED };
   State _state;

   // These bookmarks point to the current msg (not the header)
   FIXFragment::Bookmark<fix::BeginString> _begin_string_bm;
   FIXFragment::Bookmark<fix::BodyLength> _body_length_bm;
   FIXFragment::Bookmark<fix::MsgType> _msg_type_bm;

   // These bookmarks point to the header and thus have to be shifted
   // by the offset of the end of the message type.
   FIXFragment::Bookmark<fix::MsgSeqNum> _seq_num_bm;
   FIXFragment::Bookmark<fix::LastSeqNumProcessed> _last_seq_num_bm;
   typedef StringField<fix::SendingTime::tag> SendingTimeStr;
   FIXFragment::Bookmark<SendingTimeStr> _sending_time_bm;

   inline void tidy_last_msg() {
      if (_message_map.size()) {
         auto &this_msg_map = _message_map.back();
         auto start = _msg_type_bm.offset - fix::MsgType::tag_sz - 1;
         auto end = _buffer.size();
         _current_msg.overwrite(_body_length_bm, end - start, 6);
         this_msg_map.checksum = _current_msg.checksum();
         this_msg_map.checksum_bm = _current_msg.add_checksum(0);
         this_msg_map.size = _buffer.size() - this_msg_map.offset;
      }
   }

   uint32_t encode_message_type(MessageTypes m, char out[4]) {
      auto mt = (uint32_t)m;
      for (int i = 0; i < 4; i++) {
         out[i] = mt >> (8 * (3 - i));
         mt &= ((1 << (8 * (3 - i))) - 1);
      }
      for (int i = 0; i < 4; i++)
         if (out[i])
            return i;
      return 0;
   }
};

} // namespace jetfix
} // namespace bts