Walk a run of messages¶

Messages packed one after the other, each with its own length: rbe::frames::many walks them, and the adaptors in rbe::views filter them before a single field is decoded.

Source: example/howto/many.cpp, CMake target howto_many.

Declare the run¶

The element of a many must know where it ends on its own: a header with =rbe::frame_length or =rbe::payload_length, a candidate set whose id implies the size, or a fixed size. The enclosing header may bound the run with =rbe::payload_count; without it the run goes to the end of the buffer.

enum class msg_type : std::uint8_t { order = 1, cancel = 2 };

// every message carries its own length, so a run of them can be walked message by message
struct [[=rbe::pack_be]] MessageHeader {
  [[=rbe::frame_length]] std::uint8_t length;
  [[=rbe::id]]           msg_type     type;
};

struct [[=rbe::pack_be, =rbe::id(msg_type::order)]]  Order  { std::uint32_t order_id; std::uint64_t price; };
struct [[=rbe::pack_be, =rbe::id(msg_type::cancel)]] Cancel { std::uint32_t order_id; };

using message = rbe::frames::frame<MessageHeader, rbe::frames::any<Order, Cancel>>;

// the packet header says how many messages follow
struct [[=rbe::pack_be]] PacketHeader {
  std::uint32_t                        sequence;
  [[=rbe::payload_count]] std::uint8_t count;
};

using packet = rbe::frames::frame<PacketHeader, rbe::frames::many<message>>;

Iterate¶

A frame decomposes into header and payload with a structured binding, and the payload here is the run.

auto [header, messages] = rbe::deserialize<packet>(datagram);   // a frame decomposes into header and payload

std::println("packet {}: {} messages", header.field<"sequence">(), header.field<"count">());

for (auto const [msg_header, payload]: messages) {               // each element is a message frame
  payload.match(
      [](Order const& order) { std::println("  order {} @ {}", order.order_id, order.price); },
      [](Cancel const& cancel) { std::println("  cancel {}", cancel.order_id); },
      [](rbe::unmatched auto) { std::println("  unknown message"); }
  );
}

Filter¶

auto pkt = rbe::deserialize<packet>(datagram);

// a many is single-pass: iterating it advances it. payload() hands out a fresh one each time, so start
// every pipeline from it. The adaptors compose with std::views.
auto cancels = std::ranges::distance(pkt.payload() | rbe::views::with_ids(msg_type::cancel));  // some ids
auto known   = std::ranges::distance(pkt.payload() | rbe::views::known_ids());                 // ids the code knows
auto unknown = std::ranges::distance(pkt.payload() | rbe::views::unknown_ids());               // ids it does not

for (auto const [msg_header, payload]: pkt.payload() | rbe::views::with_ids(msg_type::order)) {
  std::println("  order id {}", payload.as<Order>()->field<"order_id">());
}

A many is a single-pass range: iterating it advances it. payload() returns a fresh one each call, so start every pipeline from the frame rather than from a many you have already walked.

A bare run¶

With no packet header around the messages, pipe the bytes into a many over the message view. Iteration stops at the first message that is not whole, and remainder() is what to keep for the next read.

// A run with no packet header around it: pipe the bytes straight into a many over the message view
auto messages = bytes | rbe::views::many<message::dsrl_type>();

auto seen = std::size_t {0};
for (auto const frame: messages) {
  std::println("  {} byte message", frame.length());
  ++seen;
}

// iteration stops at the first message that is not whole; the rest is kept for the next read
bool                       partial = messages.has_seen_partial();
std::span<std::byte const> rest    = messages.remainder();

See also: Frames reference, many and range adaptors.