Write a custom frame¶

The frames the library ships find their end by a length field, a count, the id or a fixed size. A protocol that ends its messages some other way, a line feed here, needs a frame that scans for it. This is the Cboe TOP feed from the market data example: ASCII lines, one message per line, no length anywhere.

Source: example/markets/cboe.hpp, CMake target markets.

The view¶

A frame view is three steps: read the header, settle the payload's extent, build the payload view. Deriving from rbe::frames::dsrl::base_frame gives the first and third in their default form through the same building blocks the library uses (header<H>::make, try_construct_payload), and the derived class writes the one that differs.

/// One TOP message as it sits on the wire: `message` plus its terminating LF and any LF padding after it.
///
/// TOP is line-oriented, so the end of a message is readable without knowing its type: the first LF
/// after the header. `line_view` is a custom frame that keeps the default header (step 1) and the default
/// payload construction (step 3, an `any` over the line, LF included) and swaps only the extent (step 2)
/// for that scan. Two things follow: `length()` covers the terminator and the padding, so `many<line>`
/// advances line by line, and the extent does not depend on the id, so an unknown `msg_type` is yielded
/// as an `any` over its own line and iteration goes on past it.
class line_view : public rbe::frames::dsrl::base_frame<Header, messages::dsrl_type> {
  using base = rbe::frames::dsrl::base_frame<Header, messages::dsrl_type>;

public:
  /// The scan for the terminator is not detectable from the shape, and the shape would get it wrong. For
  /// this case is necessary to declare the delimiting concept explicitly.
  using delimiting_concept = rbe::frames::self_delimited_tag;

  static constexpr auto make(buffer_type const buf, std::size_t const scanned_offset = 0) -> std::optional<line_view> {
    auto const hdr = header_return_type::make(buf); // step 1
    if (not hdr.has_value()) {
      return std::nullopt;
    }
    auto const rest   = buf.subspan(hdr->length());
    auto const extent = line_extent(rest, scanned_offset); // step 2: the only custom part
    if (not extent.has_value()) {
      return std::nullopt; // incomplete line: wait for more bytes
    }
    auto const line = rest.first(*extent);
    return rbe::frames::dsrl::try_construct_payload<payload_type>(*hdr, line) // step 3
        .transform([&](payload_return_type const p) { return line_view {*hdr, p, hdr->length() + line.size()}; });
  }

  /// precondition: buf holds a whole line
  constexpr explicit line_view(buffer_type const buf) : line_view(*make(buf)) { }

  [[nodiscard]] constexpr auto length() const -> size_type { return line_length_; }

private:
  /// up to and including the terminating LF, then every LF that follows it; nullopt if the line is not complete yet
  static constexpr auto line_extent(buffer_type const rest, std::size_t const scanned_offset = 0)
      -> std::optional<size_type> {
    auto const safe_offset = std::min(scanned_offset, rest.size());
    auto const unsearched  = rest.subspan(safe_offset);
    auto const lf          = std::ranges::find(unsearched, std::byte {'\n'});

    if (lf == unsearched.end()) {
      return std::nullopt;
    }

    // Calculate total extent from the beginning of 'rest', not from 'unsearched'
    return std::make_optional(std::ranges::distance(rest.begin(), lf) + 1);
  }

  constexpr line_view(header_return_type const hdr, payload_return_type const p, size_type const n) :
    base(hdr, p), line_length_(n) { }

  size_type line_length_;
};

What the derived class must provide:

static make(buffer) -> std::optional<view>: the checking constructor, which the library calls to walk a run and to implement rbe::try_deserialize.
using delimiting_concept = ...: which delimiting category the frame is in. The library cannot deduce it for an extent it does not compute: self_delimited_tag says "the frame finds its own end".
length(): the frame's extent in bytes, if it can be answered in O(1). It is what lets a many advance from one line to the next.

Everything else, header(), payload(), flatten(), the spans and the structured binding, comes from base_frame.

The vocabulary type¶

The view is what runs; the vocabulary type is what composes. It names the header, the payload, the category and the view, and that is enough for rbe::frames::many<line> and rbe::deserialize<line> to work.

/// The vocabulary serder for `line_view`: what makes `rbe::frames::many<line>` compose, exactly as `rbe::frames::blob`
/// or `rbe::frames::frame` do. It keeps the header-plus-payload shape, so the delimiting concepts classify it.
struct line {
  using header_type        = Header;
  using payload_type       = messages;
  using delimiting_concept = rbe::frames::self_delimited_tag; ///< the terminator scan of line_view, see above
  using dsrl_type          = line_view;
};

/// A run of TOP lines, LF padding included.
using lines = rbe::frames::many<line>;

Using it¶

/// Cboe TOP: a line-oriented ASCII feed. `line` is a custom frame that delimits by the LF, so `many<line>` walks
/// the stream line by line and an unknown message type is yielded as an unknown id and iteration goes on.
void cboe_top_lines() {
  using namespace cboe::top;

  auto const stream = wire(
      Header {.msg_type = message_type_t::seconds}, Seconds {.seconds = {'3', '6', '0', '0', '0'}}, std::array {'\n'},
      Header {.msg_type = static_cast<message_type_t>('?')}, std::array {'x', 'y', '\n'},
      Header {.msg_type = message_type_t::milliseconds}, Milliseconds {.milliseconds = {'0', '4', '2'}},
      std::array {'\n'}
  );

  std::println("cboe top:");
  for (auto const [header, payload]: stream | rbe::views::many<line_view>()) {
    payload.match( // clang-format off
        [](proxy<Seconds> const s) { std::println("  seconds {}", as_text(s.field<"seconds">().as_span())); },
        [](proxy<Milliseconds> const ms) { std::println("  milliseconds {}", as_text(ms.field<"milliseconds">().as_span())); },
        [](rbe::unmatched auto const other) { std::println("  skipped a line, id known: {}", other.known_id); }
    ); // clang-format on
  }
}

An unknown message type is not an error here: the extent does not depend on the id, so the line is yielded as an any with an unknown id and iteration moves on to the next line.