Dispatcher

The Dispatcher routes framed messages from one byte stream to per-module handlers, letting several independent protocols share a single USB vendor / CDC / socket / UART link. Rather than run a separate StreamParser per protocol over the same bytes (each re-buffering the whole stream and needing its own reset-on-overflow bookkeeping), a Dispatcher parses the Stream Frame APIs stream once and hands each complete frame to the handler registered for its module id.

Module id

The frame’s module byte (0..255) is the routing key — a full byte, so up to 256 protocols can coexist on one stream. The message/transaction type and the request/reply direction (flags) travel with the frame and are handed to the module’s handler untouched; the Dispatcher does not interpret them. espp’s own protocols and examples use these ids by default:

Module id	Protocol
0	OTA
1	Core-dump example crash trigger (example only)
2	BLDC haptics
3	Telemetry
4	crash dump
5	CAN bridge
6	MCP266 console
0xF0-0xFE	reserved (meta)
0xFF	capability discovery

A device-side dispatcher registers the modules it serves; frames for an unregistered module are silently ignored. A protocol’s replies use the same module as its requests (the reply/direction lives in the frame’s flags, not the module), so both directions route to the one registered handler — use frame.is_reply() to tell them apart. In practice a device only receives requests (it sends the replies), so its handler normally sees requests only. Application code may assign any unused module id to its own protocol; nothing is hard-wired to a specific service. The ids in the table are defaults: the module id is only a routing key, and every espp service takes its id from Config::module (used for both the requests it accepts and the replies it sends), while each example module keeps its id in one named constant. The defaults are what the hosted web consoles and the espp_ota CLI look for, so a device that moves a service off its default must also tell its host tooling the new id — see Building Custom Modules & Protocols.

The example multiplexes two toy protocols over one in-memory byte stream (the same feed() / register_module() pattern applies to a USB vendor / CDC / socket / UART receive path):

  namespace sf = espp::stream_frame;

  // Two toy protocols sharing one stream: "control" on module 0 and "telemetry"
  // on module 4 (module is a full byte, so up to 256 protocols can coexist).
  static constexpr uint8_t kModuleControl = 0;
  static constexpr uint8_t kModuleTelemetry = 4;

  espp::Dispatcher dispatcher;
  // Register each module WITH discovery metadata (name / web app / description)
  // so a connected peer can enumerate them (see the discovery section below).
  dispatcher.register_module(
      kModuleControl,
      [&](const sf::Frame &f) {
        logger.info("[control] {} type=0x{:02X} ({} payload bytes)",
                    f.is_reply() ? "reply" : "request", f.type, f.payload.size());
      },
      {.name = "Control", .app = "control_console.html", .description = "Device control channel"});
  dispatcher.register_module(kModuleTelemetry,
                             [&](const sf::Frame &f) {
                               uint32_t value = f.payload.size() == 4 ? sf::get_u32(f.payload) : 0;
                               logger.info("[telemetry] type=0x{:02X} value={}", f.type, value);
                             },
                             {.name = "Telemetry",
                              .app = "telemetry_console.html",
                              .description = "Live telemetry stream"});

  // Build a mixed stream, as a peer would send it.
  std::vector<uint8_t> telemetry_payload;
  sf::put_u32(telemetry_payload, 42);
  std::vector<uint8_t> stream;
  auto append = [&](const std::vector<uint8_t> &frame) {
    stream.insert(stream.end(), frame.begin(), frame.end());
  };
  const uint8_t start[] = {'s', 't', 'a', 'r', 't'};
  append(sf::build_frame(/*reply=*/false, kModuleControl, 0x01, start));
  append(sf::build_frame(/*reply=*/false, kModuleTelemetry, 0x01, telemetry_payload));
  // A frame for an unregistered module is silently ignored.
  append(sf::build_frame(/*reply=*/false, 9, 0x00));
  append(sf::build_frame(/*reply=*/true, kModuleControl, 0x81)); // a reply/event

  // Feed it in two arbitrary chunks to show the parser reassembles split frames.
  const size_t half = stream.size() / 2;
  dispatcher.feed(std::span<const uint8_t>(stream.data(), half));
  dispatcher.feed(std::span<const uint8_t>(stream.data() + half, stream.size() - half));
  logger.info("done ({} bytes dropped while resyncing)", dispatcher.dropped_bytes());

  // --- capability discovery -------------------------------------------------
  // A connected peer (e.g. the browser hub app) can ask WHICH modules this
  // device runs, over the reserved discovery module id 0xFF. Advertise a device
  // name + firmware, then opt in to auto-answering the query. serve_discovery()
  // is the ONLY path by which the Dispatcher sends: it hands the encoded reply
  // frame to the transmit callback we supply. On real hardware that callback is
  // usb.write_vendor / socket send / etc.; here we just capture it in-process
  // and decode it to show what a peer receives.
  dispatcher.set_device_info("espp Dispatcher Example", "1.0.0");
  std::vector<uint8_t> discovery_reply;
  dispatcher.serve_discovery(
      [&](std::span<const uint8_t> frame) { discovery_reply.assign(frame.begin(), frame.end()); });

  dispatcher.feed(sf::build_frame(/*reply=*/false, espp::Dispatcher::kDiscoveryModule,
                                  static_cast<uint8_t>(espp::Dispatcher::Discovery::ListModules)));

  const auto reply_frames = sf::StreamParser{}.feed(discovery_reply);
  if (!reply_frames.empty()) {
    const auto &p = reply_frames[0].payload;
    size_t i = 2; // skip [version][reserved]
    auto rd_str = [&]() {
      const uint8_t n = p[i++];
      std::string s(reinterpret_cast<const char *>(&p[i]), n);
      i += n;
      return s;
    };
    const std::string dev = rd_str();
    const std::string fw = rd_str();
    const uint8_t count = p[i++];
    logger.info("discovery: '{}' (fw {}) advertises {} module(s):", dev, fw, count);
    for (uint8_t m = 0; m < count; ++m) {
      const uint8_t id = p[i++];
      const std::string name = rd_str();
      const std::string app = rd_str();
      const std::string desc = rd_str();
      logger.info("  module {}: {} [app={}] — {}", id, name, app, desc);
    }
  }

The codec and dispatcher are also exposed to Python (espp.stream_frame and espp.Dispatcher); see python/dispatcher.py and python/dispatcher_test.py.

API Reference

Header File

components/dispatcher/include/dispatcher.hpp

Classes

class Dispatcher

Routes framed messages from one byte stream to per-module handlers.

Public Types

enum class Discovery : uint8_t

`type` values within the discovery module (kDiscoveryModule).

Values:

enumerator ListModules: request: list the device’s modules; reply payload = describe().

using handler_fn = std::function<void(const stream_frame::Frame &frame)>

Handler invoked for every frame whose module was registered.

Param frame:: The decoded frame (module, type, flags/reply, payload).

using reply_fn = std::function<void(std::span<const uint8_t> frame)>: Transmit callback for serve_discovery(): sends one already-encoded stream_frame back to the peer over the application’s transport.

using ModuleInfo = DispatcherModuleInfo: Discovery metadata for a module (see DispatcherModuleInfo).

Public Functions

inline void register_module(uint8_t module_id, handler_fn handler)

Register (or replace) the handler for a module id.

Parameters:

module_id – Module id (0..255) to route to handler.
handler – Callback for frames with this module id. A null handler unregisters the module.

inline void register_module(uint8_t module_id, handler_fn handler, ModuleInfo info)

Register (or replace) a module’s handler AND its discovery metadata.

Note

Registration fully replaces any previous entry, so re-registering a module with the 2-argument overload clears its metadata.

Parameters:

module_id – Module id (0..255) to route to handler.
handler – Callback for frames with this module id (a null handler unregisters the module; info is then ignored).
info – Metadata advertised to a discovery peer (see describe()).

template<DispatcherModuleConcept Service> inline void register_module(Service &service)

Register a *module object* satisfying DispatcherModuleConcept (`module_id()` / `module_info()` / `handle(frame)` — every espp protocol service does: OtaService, CoreDumpService, Telemetry, Mcp266Service, …). Equivalent to registering `[&](auto &f) { service.handle(f); }` on `service.module_id()` with `service.module_info()`.

The id and metadata are read from the *object*, so a service whose module id is configured per instance (e.g. an app module constructed with `{.module = 0x20}`) registers under that id; a class that only provides `static` members works too, since a static member resolves through an object expression as well.

Parameters:: service – The service; must outlive its registration.

inline void unregister_module(uint8_t module_id): Remove the handler for a module id (frames for it become ignored).

inline bool has_module(uint8_t module_id) const: Whether a handler is registered for a module id (reflects applied registrations; changes made during a dispatch apply after it ends).

inline void set_device_info(std::string name, std::string firmware = ""): Set the device-level info advertised at the head of describe() (so a peer can show “Connected to <name> <firmware>”).

inline std::vector<uint8_t> describe() const

Serialize the device info + every registered module that carries a (non-empty) name into the binary discovery payload.

Layout (all lengths are one byte; strings are [len][bytes], truncated at 255): [version u8][reserved u8][device_name str][device_fw str] [module_count u8] then per module [id u8][name str][app str][desc str]. The reserved discovery module (0xFF) is never listed. At most 255 modules are emitted, and trailing modules are dropped if the payload would exceed stream_frame::kMaxPayloadSize — module_count always reflects the number actually emitted, so the payload is self-consistent and fits one frame.

Returns:: The payload bytes (to be sent as the ListModules reply).

inline void serve_discovery(reply_fn reply)

Opt in to auto-answering capability discovery.

Registers a handler on kDiscoveryModule that, on a Discovery::ListModules request, encodes describe() into a reply frame (echoing the request’s correlation id, if any) and hands it to reply for transmission. This is the only path by which a Dispatcher ever sends — the app supplies the transport.

Parameters:: reply – Transmit callback (sends the encoded reply frame).

inline void feed(std::span<const uint8_t> data): Feed raw received bytes: parse and route each complete frame to its module’s handler. Frames whose module has no handler are ignored (so unrelated protocols on the same stream are harmless).

inline void dispatch(const stream_frame::Frame &frame): Route an already-parsed frame (for callers running their own parser).

inline void reset(): Discard any partially-buffered frame bytes (transport reconnect or RX overflow) so a frame straddling the gap resynchronizes at once.

inline size_t buffered() const: Bytes buffered awaiting frame completion.

inline size_t dropped_bytes() const: Total bytes discarded while resynchronizing (diagnostics).

Public Static Functions

static inline constexpr uint8_t module_of(const stream_frame::Frame &frame): The module id a frame will route to (its `module` byte).

Public Static Attributes

static constexpr uint8_t kDiscoveryModule = 0xFF: Reserved module id for capability discovery. A peer sends a Discovery::ListModules request here; the device answers with the serialized module list (see describe() / serve_discovery()). Module ids 0xF0..0xFF are reserved for dispatcher / meta use.

static constexpr uint8_t kDiscoveryVersion = 1: Version byte at the start of a describe() payload, so the wire format can evolve without a framing change.

Header File

components/dispatcher/include/dispatcher_worker.hpp

Classes

class DispatcherWorker : public espp::BaseComponent 

A Dispatcher plus the bounded receive queue and worker task that feed it.

Public Types

using send_fn = Dispatcher::reply_fn : Transmit callback: sends one encoded frame over this worker’s transport.

using overflow_fn = std::function<void()>: Called on the worker task after an RX overflow was handled (queue cleared, parser reset).

Public Functions

inline explicit DispatcherWorker(const Config &config): Construct and start the worker task.

inline ~DispatcherWorker(): Stop the worker task (queued bytes are discarded).

inline send_fn sender() const: The transmit function configured for this transport (for services that reply on the stream they were registered on).

template<DispatcherModuleConcept Service> inline void register_module(Service &service): Register a protocol service satisfying DispatcherModuleConcept (see Dispatcher::register_module(Service&)).

inline void register_module(uint8_t module_id, Dispatcher::handler_fn handler, Dispatcher::ModuleInfo info = {}): Register a raw handler for a module id (see Dispatcher::register_module).

inline void unregister_module(uint8_t module_id): Remove a module’s handler (see Dispatcher::unregister_module).

inline void serve_discovery(std::string device_name, std::string firmware = "")

Advertise the device and answer capability discovery (module 0xFF) over this transport’s `send` (see Dispatcher::serve_discovery).

Parameters:

device_name – Device name shown by discovery peers.
firmware – Firmware version string (optional).

inline Dispatcher &dispatcher(): Direct access to the underlying Dispatcher. Only safe from a module handler (which runs on the worker) or before any bytes are pushed; prefer the register_module()/serve_discovery() wrappers otherwise.

inline bool push(std::span<const uint8_t> data)

Queue received bytes for the worker. Callable from any task (e.g. a transport’s receive callback); never blocks beyond a short lock.

Returns:: false if the bytes were dropped (overflow).

inline void request_reset(): Ask the worker to discard any half-parsed frame before the next bytes (transport connect / disconnect). Bytes still queued from before the reset are dropped too — they belong to the old link and must not be fed after the parser reset. Safe from any task.

inline size_t queued_bytes() const: Bytes currently queued for the worker.

inline size_t overflows() const: Total overflow events (diagnostics).

inline const std::string &get_name() const

Get the name of the component

Note

This is the tag of the logger

Returns:: A const reference to the name of the component

inline void set_log_tag(const std::string_view &tag)

Set the tag for the logger

Parameters:: tag – The tag to use for the logger

inline espp::Logger::Verbosity get_log_level() const

Get the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Returns:: The verbosity level of the logger

inline void set_log_level(espp::Logger::Verbosity level)

Set the log level for the logger

See also

Logger::Verbosity

See also

Logger::set_verbosity

Parameters:: level – The verbosity level to use for the logger

inline void set_log_verbosity(espp::Logger::Verbosity level)

Set the log verbosity for the logger

See also

set_log_level

See also

Logger::Verbosity

See also

Logger::set_verbosity

Note

This is a convenience method that calls set_log_level

Parameters:: level – The verbosity level to use for the logger

inline espp::Logger::Verbosity get_log_verbosity() const

Get the log verbosity for the logger

See also

get_log_level

See also

Logger::Verbosity

See also

Logger::get_verbosity

Note

This is a convenience method that calls get_log_level

Returns:: The verbosity level of the logger

inline void set_log_rate_limit(std::chrono::duration<float> rate_limit)

Set the rate limit for the logger

See also

Logger::set_rate_limit

Note

Only calls to the logger that have _rate_limit suffix will be rate limited

Parameters:: rate_limit – The rate limit to use for the logger

struct Config

Configuration for the DispatcherWorker.

Public Members

send_fn send = {nullptr}: Sends one encoded frame over this transport. Used by serve_discovery() and returned by sender() so services can share it; optional if the worker only routes.

overflow_fn on_overflow = {nullptr}: Notified on the worker task after an overflow (see the header comment).

size_t max_queued_bytes = {8 * stream_frame::kMaxFrameSize}: Bound on queued-but-unprocessed bytes; a push that would exceed it drops everything queued and flags an overflow.

Task::BaseConfig task_config = {.name = "dispatcher", .stack_size_bytes = 8192}: Worker task settings (name / stack / priority / core). Size the stack for the handlers it runs (OTA + logging comfortably fit 8 KiB).

espp::Logger::Verbosity log_level = {espp::Logger::Verbosity::WARN}: Logger verbosity.