Skip to content

feat: RELAY spec v0.2 conformance — relay:: namespace, Channel<T>, Adapt() #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SoundMatt merged 1 commit into from
Jun 17, 2026
Merged

feat: RELAY spec v0.2 conformance — relay:: namespace, Channel<T>, Adapt() #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
203 changes: 203 additions & 0 deletions include/rcp/adapt.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,203 @@
// fusa:req REQ-RELAY-001
// fusa:req REQ-RELAY-002
// fusa:req REQ-RELAY-003
// fusa:req REQ-RELAY-004
// fusa:req REQ-RELAY-005

// RELAY application interface adapter for cpp-RCP (§10.3, §18.2).
//
// Adapt() wraps an rcp::Controller as a relay::Caller so application code can
// use the protocol-agnostic relay::Node / relay::Caller interface and swap the
// underlying protocol with a single constructor change.
//
// Usage:
// auto ctrl = std::make_shared<rcp::mock::Controller>(rcp::Zone::FrontLeft);
// auto caller = rcp::Adapt(ctrl); // relay::Caller*
// auto [resp, ec] = caller->call(ctx, msg);
#pragma once

#include <rcp/rcp.hpp>
#include <relay/relay.hpp>

#include <charconv>
#include <chrono>
#include <memory>
#include <string>
#include <thread>

namespace rcp {

// ── Zone ↔ relay::Message.ID helpers (§15.7.5) ───────────────────────────────

// zone_to_relay_id returns the RELAY-canonical PascalCase zone name (§4.2).
inline std::string zone_to_relay_id(Zone z) {
switch (z) {
case Zone::FrontLeft: return "FrontLeft";
case Zone::FrontRight: return "FrontRight";
case Zone::RearLeft: return "RearLeft";
case Zone::RearRight: return "RearRight";
case Zone::Central: return "Central";
default: return "Unknown";
}
}

// zone_from_relay_id parses a PascalCase zone name from relay::Message.ID.
inline Zone zone_from_relay_id(const std::string& id) {
if (id == "FrontLeft") return Zone::FrontLeft;
if (id == "FrontRight") return Zone::FrontRight;
if (id == "RearLeft") return Zone::RearLeft;
if (id == "RearRight") return Zone::RearRight;
if (id == "Central") return Zone::Central;
return Zone::Unknown;
}

// priority_from_meta parses the "rcp.priority" meta key.
inline Priority priority_from_meta(const std::map<std::string, std::string>& meta) {
auto it = meta.find("rcp.priority");
if (it == meta.end()) return Priority::Normal;
if (it->second == "high") return Priority::High;
if (it->second == "critical") return Priority::Critical;
return Priority::Normal;
}

// cmd_type_from_meta parses the "rcp.cmd_type" meta key.
inline CommandType cmd_type_from_meta(const std::map<std::string, std::string>& meta) {
auto it = meta.find("rcp.cmd_type");
if (it == meta.end()) return CommandType::Noop;
const auto& s = it->second;
if (s == "set") return CommandType::Set;
if (s == "get") return CommandType::Get;
if (s == "reset") return CommandType::Reset;
if (s == "watchdog") return CommandType::Watchdog;
if (s == "sleep") return CommandType::Sleep;
if (s == "wake") return CommandType::Wake;
return CommandType::Noop;
}

// ── ToMessage / FromMessage (§15.7.5) ────────────────────────────────────────

// status_to_message converts rcp::Status to relay::Message (subscribe direction).
inline relay::Message status_to_message(const Status& s) {
relay::Message msg;
msg.protocol = relay::Protocol::RCP;
msg.id = zone_to_relay_id(s.zone);
msg.payload = s.payload;
msg.seq = s.seq;
msg.timestamp = std::chrono::system_clock::now();
msg.meta["rcp.healthy"] = s.healthy ? "true" : "false";
return msg;
}

// response_to_message converts rcp::Response to relay::Message (call direction).
inline relay::Message response_to_message(const Response& r) {
relay::Message msg;
msg.protocol = relay::Protocol::RCP;
msg.id = zone_to_relay_id(r.zone);
msg.payload = r.payload;
msg.timestamp = std::chrono::system_clock::now();
msg.meta["rcp.status"] = std::to_string(static_cast<int>(r.status));
return msg;
}

// message_to_command converts a relay::Message to rcp::Command (call direction).
inline Command message_to_command(const relay::Message& msg) {
Command cmd;
cmd.zone = zone_from_relay_id(msg.id);
cmd.priority = priority_from_meta(msg.meta);
cmd.type = cmd_type_from_meta(msg.meta);
cmd.payload = msg.payload;
return cmd;
}

// ── RcpCallerAdapter — implements relay::Caller over rcp::Controller ──────────

class RcpCallerAdapter final : public relay::Caller {
public:
explicit RcpCallerAdapter(std::shared_ptr<Controller> ctrl)
: ctrl_(std::move(ctrl)) {}

relay::Protocol protocol() const noexcept override {
return relay::Protocol::RCP;
}

// send maps relay::Message → rcp::Command, discards the response (§10.6).
std::error_code send(relay::Context ctx, const relay::Message& msg) override {
if (!ctrl_) return make_error_code(Errc::not_found);
Command cmd = message_to_command(msg);
cmd.zone = zone_from_relay_id(msg.id);
Response resp;
return ctrl_->send(ctx, cmd, resp);
}

// call maps relay::Message → rcp::Command → relay::Message (§10.2).
std::pair<relay::Message, std::error_code>
call(relay::Context ctx, const relay::Message& req) override {
if (!ctrl_) return {{}, make_error_code(Errc::not_found)};
Command cmd = message_to_command(req);
Response resp;
auto ec = ctrl_->send(ctx, cmd, resp);
if (ec) return {{}, ec};
return {response_to_message(resp), {}};
}

// subscribe wraps rcp::Controller::subscribe, forwarding Status as relay::Message.
// Starts a background thread per §10.5 that reads from StatusChannel and pushes
// to the relay::Channel<Message>. Thread exits when the status channel closes.
std::pair<std::shared_ptr<relay::Channel<relay::Message>>, std::error_code>
subscribe(relay::SubscriberOptions opts = {}) override {
if (!ctrl_) return {nullptr, make_error_code(Errc::not_found)};

std::shared_ptr<StatusChannel> status_ch;
auto ec = ctrl_->subscribe(relay::Context::background(), status_ch);
if (ec) return {nullptr, ec};

auto out = std::make_shared<relay::Channel<relay::Message>>(opts.channel_depth);
auto bp = opts.back_pressure;

std::thread([status_ch, out, bp]() mutable {
while (true) {
auto st = status_ch->recv();
if (!st) break;
auto msg = status_to_message(*st);
switch (bp) {
case relay::BackPressurePolicy::drop_newest:
out->push(std::move(msg));
break;
case relay::BackPressurePolicy::drop_oldest:
if (!out->push(msg)) {
out->try_recv();
out->push(std::move(msg));
}
break;
case relay::BackPressurePolicy::block:
while (!out->push(msg) && !out->is_closed()) {
// spin until space available or channel closed
}
break;
}
}
out->close();
}).detach();

return {out, {}};
}

std::error_code close() noexcept override {
if (!ctrl_) return {};
return ctrl_->close();
}

private:
std::shared_ptr<Controller> ctrl_;
};

// ── Adapt() (§10.3) ──────────────────────────────────────────────────────────

// Adapt wraps an rcp::Controller as a relay::Caller.
// The returned relay::Caller also satisfies relay::Node.
// Does NOT block or connect; wraps the given controller immediately.
inline std::unique_ptr<relay::Caller> Adapt(std::shared_ptr<Controller> ctrl) {
return std::make_unique<RcpCallerAdapter>(std::move(ctrl));
}

} // namespace rcp
107 changes: 26 additions & 81 deletions include/rcp/rcp.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,16 +48,18 @@
// A central high-performance computer uses a Registry to discover zone controllers,
// then dispatches Commands to each Controller and receives Responses and Status
// telemetry in return.
//
// RELAY conformance: include <relay/relay.hpp> for relay:: namespace types, and
// <rcp/adapt.hpp> for Adapt() which wraps a Controller as a relay::Caller.
#pragma once

#include <relay/relay.hpp>

#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <string>
#include <system_error>
Expand Down Expand Up @@ -90,6 +92,20 @@ inline const std::error_category& rcp_category() noexcept {
default: return "rcp: unknown error";
}
}
// Map rcp::Errc codes to relay::Errc conditions (§5.2, §5.3).
bool equivalent(int code, const std::error_condition& cond) const noexcept override {
if (cond.category() != relay::relay_category()) return false;
auto re = static_cast<relay::Errc>(cond.value());
switch (static_cast<Errc>(code)) {
case Errc::closed: return re == relay::Errc::closed;
case Errc::timeout: return re == relay::Errc::timeout;
case Errc::busy: return re == relay::Errc::timeout;
case Errc::not_found: return re == relay::Errc::not_connected;
case Errc::zone_mismatch: return re == relay::Errc::not_connected;
case Errc::already_exists: return re == relay::Errc::closed; // §5.3; see RELAY issue #8
default: return false;
}
}
};
static Cat instance;
return instance;
Expand Down Expand Up @@ -225,87 +241,16 @@ class Loan {
std::function<void()> release_;
};

// ── Context ───────────────────────────────────────────────────────────────────
// Lightweight context carrying an optional deadline. Analogous to Go's context.Context.

class Context {
public:
static Context background() noexcept { return Context{}; }

static Context with_deadline(std::chrono::steady_clock::time_point tp) noexcept {
Context c;
c.deadline_ = tp;
return c;
}

static Context with_timeout(std::chrono::steady_clock::duration d) noexcept {
return with_deadline(std::chrono::steady_clock::now() + d);
}

bool done() const noexcept {
if (!deadline_) return false;
return std::chrono::steady_clock::now() >= *deadline_;
}

std::optional<std::chrono::steady_clock::time_point> deadline() const noexcept {
return deadline_;
}

private:
std::optional<std::chrono::steady_clock::time_point> deadline_;
};

// ── StatusChannel ─────────────────────────────────────────────────────────────
// Thread-safe bounded channel for Status updates, analogous to a Go channel.
// ── Context — relay::Context alias (§18.2) ────────────────────────────────────
// rcp::Context is an alias for relay::Context per §18.2 conformance.

class StatusChannel {
public:
explicit StatusChannel(size_t cap = 16) : cap_(cap) {}

void push(const Status& s) {
std::lock_guard<std::mutex> lk(mu_);
if (closed_) return;
if (queue_.size() < cap_) {
queue_.push_back(s);
cv_.notify_one();
}
}
using Context = relay::Context;

std::optional<Status> recv() {
std::unique_lock<std::mutex> lk(mu_);
cv_.wait(lk, [this] { return !queue_.empty() || closed_; });
if (queue_.empty()) return std::nullopt;
auto s = queue_.front();
queue_.pop_front();
return s;
}
// ── StatusChannel — relay::Channel<Status> alias (§18.2) ─────────────────────
// rcp::StatusChannel is an alias for relay::Channel<Status>.
// Legacy code using std::shared_ptr<StatusChannel> is unchanged.

std::optional<Status> try_recv() {
std::lock_guard<std::mutex> lk(mu_);
if (queue_.empty()) return std::nullopt;
auto s = queue_.front();
queue_.pop_front();
return s;
}

void close() {
std::lock_guard<std::mutex> lk(mu_);
closed_ = true;
cv_.notify_all();
}

bool is_closed() const noexcept {
std::lock_guard<std::mutex> lk(mu_);
return closed_;
}

private:
mutable std::mutex mu_;
std::condition_variable cv_;
std::deque<Status> queue_;
size_t cap_;
bool closed_ = false;
};
using StatusChannel = relay::Channel<Status>;

// ── Controller ────────────────────────────────────────────────────────────────

Expand Down
Loading
Loading