raphael/secs-gem
Public Access
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

hsms: HSMS-GS multi-session support (E37 §11)

Browse Source 
Connection now supports both HSMS-SS (single session — the
constructor's behaviour, unchanged) and HSMS-GS (multi-session).
add_session(device_id) registers additional sessions; each one has
its own NotSelected/Selected state and its own message/selected
handlers.  In GS mode the Select.req carries session_id=device_id;
in SS mode it stays at 0xFFFF (legacy).  Linktest/Separate remain
connection-scope per spec.

Public API additions:
  add_session(device_id)
  set_session_message_handler(device_id, h)
  set_session_selected_handler(device_id, h)
  session_state(device_id) -> State
  is_session_selected(device_id) -> bool
  send_request(device_id, msg, cb)
  send_data(device_id, msg)

Internal refactor: state_/on_message_/on_selected_ folded into a
SessionSlot map keyed by device_id; SS-style getters/setters route
through the primary session.  T7 + linktest are connection-scope —
T7 fires only when no session is selected; linktest runs while at
least one is.

Five wire-level tests:
  - passive: two sessions selected independently via Select.req
    with their own session_id
  - GS Select.req for an unregistered session id is Rejected
    (EntityNotSelected)
  - data routed by session_id; data on a not-selected session is
    Rejected
  - active: two registered sessions both end up selected via
    serialized Select.req per session
  - SS legacy: existing single-session API still works (session_id
    0xFFFF in Select.req)

COMPLIANCE.md §1 updated: HSMS-GS row goes ⬜ → ✅.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Raphael Maenle
2026-06-09 11:40:38 +02:00
parent 998e81b3d8
commit d0c7fb71b6
5 changed files with 516 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/hsms/connection.cpp
+147 -37
View File
@@ -39,12 +39,66 @@ Connection::Connection(asio::ip::tcp::socket socket, Mode mode, uint16_t device_
linktest_timer_(socket_.get_executor()),
mode_(mode),
device_id_(device_id),
timers_(timers) {}
timers_(timers) {
SessionSlot primary;
primary.device_id = device_id;
sessions_.emplace(device_id, std::move(primary));
}
void Connection::set_message_handler(MessageHandler h) {
if (auto* s = find_session(device_id_)) s->on_message = std::move(h);
}
void Connection::set_selected_handler(SelectedHandler h) {
if (auto* s = find_session(device_id_)) s->on_selected = std::move(h);
}
Connection::State Connection::state() const {
auto* s = find_session(device_id_);
return s ? s->state : State::NotSelected;
}
bool Connection::selected() const { return state() == State::Selected; }
void Connection::add_session(uint16_t device_id) {
if (sessions_.count(device_id)) return;
SessionSlot s;
s.device_id = device_id;
sessions_.emplace(device_id, std::move(s));
if (mode_ == Mode::Active) {
pending_active_selects_.push_back(device_id);
}
}
void Connection::set_session_message_handler(uint16_t device_id, MessageHandler h) {
if (auto* s = find_session(device_id)) s->on_message = std::move(h);
}
void Connection::set_session_selected_handler(uint16_t device_id, SelectedHandler h) {
if (auto* s = find_session(device_id)) s->on_selected = std::move(h);
}
Connection::State Connection::session_state(uint16_t device_id) const {
auto* s = find_session(device_id);
return s ? s->state : State::NotSelected;
}
bool Connection::is_session_selected(uint16_t device_id) const {
return session_state(device_id) == State::Selected;
}
Connection::SessionSlot* Connection::resolve_select_target(uint16_t sid) {
if (gs_mode()) return find_session(sid);
// SS legacy: session_id is typically 0xFFFF in Select.req but some
// peers also send the device_id. Either way, route to the primary.
return find_session(device_id_);
}
Connection::SessionSlot* Connection::resolve_data_target(uint16_t sid) {
if (gs_mode()) return find_session(sid);
return find_session(device_id_);
}
void Connection::start() {
read_length();
if (mode_ == Mode::Active) {
send_select_req();
if (gs_mode()) send_select_req(device_id_);
else send_select_req();
} else {
arm_t7();
}
@@ -167,9 +221,8 @@ void Connection::handle_data(const Frame& frame) {
const Header& h = frame.header;
// Reply correlation: match on (system_bytes, stream, function) exactly.
// SECS-II replies use the same system_bytes as their request and have
// stream==request.stream, function==request.function+1; we recorded both
// when send_request stored the pending transaction.
// System bytes are connection-scope (E37 §8.3); replies route by system
// bytes regardless of which session they came in on.
auto it = pending_requests_.find(h.system_bytes);
if (it != pending_requests_.end() &&
it->second.expected_stream == h.stream() &&
@@ -208,8 +261,10 @@ void Connection::handle_data(const Frame& frame) {
// handler can deal with it; the pending request stays open until T3.
}
// Primary message; only valid while SELECTED.
if (state_ != State::Selected) {
// Primary message; route by header.session_id (GS) or to primary (SS).
// Reject if no matching session OR if that session isn't SELECTED.
SessionSlot* target = resolve_data_target(h.session_id);
if (!target || target->state != State::Selected) {
send_frame(Frame(Header::control(SType::RejectReq, h.system_bytes, kControlSessionId,
static_cast<uint8_t>(SType::Data),
static_cast<uint8_t>(RejectReason::EntityNotSelected))));
@@ -228,16 +283,16 @@ void Connection::handle_data(const Frame& frame) {
}
log("<- " + h.describe());
if (!on_message_) return;
if (!target->on_message) return;
// Expose the originating header to the handler in case it needs to emit
// an S9F3 / S9F5 in response. Cleared on the way out.
current_header_ = &h;
auto reply = on_message_(msg);
auto reply = target->on_message(msg);
current_header_ = nullptr;
if (reply) {
Frame out(Header::data_message(device_id_, reply->stream, reply->function,
Frame out(Header::data_message(target->device_id, reply->stream, reply->function,
reply->reply_expected, h.system_bytes),
reply->encode_body());
log("-> " + out.header.describe());
@@ -249,10 +304,18 @@ void Connection::handle_control(const Frame& frame) {
const Header& h = frame.header;
switch (h.stype) {
case SType::SelectReq: {
// E37 §7.2: a Select.req while already SELECTED is answered with
// SelectStatus::AlreadyActive; we do NOT transition (we're already
// there) and we do NOT call the selected handler twice.
const auto status = (state_ == State::Selected)
// Route by header.session_id (GS) or to primary (SS legacy).
SessionSlot* tgt = resolve_select_target(h.session_id);
if (!tgt) {
// GS: no such session on this connection.
log("<- Select.req for unknown session " + std::to_string(h.session_id));
send_frame(Frame(Header::control(
SType::RejectReq, h.system_bytes, h.session_id,
static_cast<uint8_t>(SType::SelectReq),
static_cast<uint8_t>(RejectReason::EntityNotSelected))));
break;
}
const auto status = (tgt->state == State::Selected)
? SelectStatus::AlreadyActive
: SelectStatus::Ok;
log(std::string("<- Select.req") +
@@ -261,7 +324,7 @@ void Connection::handle_control(const Frame& frame) {
static_cast<uint8_t>(status))));
log(std::string("-> Select.rsp (") +
(status == SelectStatus::Ok ? "Ok" : "AlreadyActive") + ")");
if (status == SelectStatus::Ok) enter_selected();
if (status == SelectStatus::Ok) enter_selected(tgt->device_id);
break;
}
case SType::SelectRsp: {
@@ -270,7 +333,16 @@ void Connection::handle_control(const Frame& frame) {
clear_control_transaction();
if (h.byte3 == static_cast<uint8_t>(SelectStatus::Ok)) {
log("<- Select.rsp (Ok)");
enter_selected();
// Identify which session this rsp completes by session_id; SS
// legacy mode uses 0xFFFF on the wire, default to primary.
uint16_t selected = gs_mode() ? h.session_id : device_id_;
enter_selected(selected);
// Active mode walk-list: kick off next pending select.
if (mode_ == Mode::Active && !pending_active_selects_.empty()) {
uint16_t next = pending_active_selects_.front();
pending_active_selects_.pop_front();
send_select_req(next);
}
} else {
close("Select rejected, status=" + std::to_string(h.byte3));
}
@@ -278,18 +350,20 @@ void Connection::handle_control(const Frame& frame) {
break;
}
case SType::DeselectReq: {
// E37 §7.4: Deselect.req while NOT_SELECTED returns NotEstablished
// and leaves state untouched.
const auto status = (state_ == State::Selected)
SessionSlot* tgt = resolve_select_target(h.session_id);
const auto status = (tgt && tgt->state == State::Selected)
? DeselectStatus::Ok
: DeselectStatus::NotEstablished;
log(std::string("<- Deselect.req") +
(status == DeselectStatus::NotEstablished ? " (not SELECTED)" : ""));
send_frame(Frame(Header::control(SType::DeselectRsp, h.system_bytes, h.session_id, 0,
static_cast<uint8_t>(status))));
if (status == DeselectStatus::Ok) {
state_ = State::NotSelected;
arm_t7();
if (status == DeselectStatus::Ok && tgt) {
tgt->state = State::NotSelected;
// Re-arm T7 only if NO sessions are selected.
bool any_selected = false;
for (auto& kv : sessions_) if (kv.second.state == State::Selected) { any_selected = true; break; }
if (!any_selected) arm_t7();
}
break;
}
@@ -297,7 +371,8 @@ void Connection::handle_control(const Frame& frame) {
if (pending_control_ && pending_control_->expected_response == SType::DeselectRsp &&
pending_control_->system_bytes == h.system_bytes) {
clear_control_transaction();
state_ = State::NotSelected;
uint16_t target = gs_mode() ? h.session_id : device_id_;
if (auto* s = find_session(target)) s->state = State::NotSelected;
}
break;
}
@@ -357,32 +432,33 @@ void Connection::write_next() {
// --- public send API ------------------------------------------------------
void Connection::send_request(secs2::Message msg, ReplyHandler cb) {
send_request(device_id_, std::move(msg), std::move(cb));
}
void Connection::send_request(uint16_t device_id, secs2::Message msg, ReplyHandler cb) {
if (closed_) {
cb(make_error(Error::Closed), {});
return;
}
const uint32_t sys = next_system_bytes();
Frame f(Header::data_message(device_id_, msg.stream, msg.function, true, sys),
Frame f(Header::data_message(device_id, msg.stream, msg.function, true, sys),
msg.encode_body());
auto t3 = std::make_shared<asio::steady_timer>(socket_.get_executor());
t3->expires_after(timers_.t3);
// SECS-II reply convention: same stream, function+1.
pending_requests_.emplace(sys, PendingRequest{
/*expected_stream=*/msg.stream,
/*expected_function=*/static_cast<uint8_t>(msg.function + 1),
std::move(cb), t3});
auto self = shared_from_this();
t3->async_wait([this, self, sys](std::error_code ec) {
t3->async_wait([this, self, sys, device_id](std::error_code ec) {
if (ec) return;
auto it = pending_requests_.find(sys);
if (it == pending_requests_.end()) return;
// Reconstruct the MHEAD of the request that timed out and notify the
// peer with S9F9 before surfacing Timeout to the caller.
const uint8_t req_stream = it->second.expected_stream;
const uint8_t req_function = static_cast<uint8_t>(it->second.expected_function - 1);
Header h = Header::data_message(device_id_, req_stream, req_function, true, sys);
Header h = Header::data_message(device_id, req_stream, req_function, true, sys);
emit_s9(9, h.encode());
auto cb2 = std::move(it->second.cb);
pending_requests_.erase(it);
@@ -394,9 +470,13 @@ void Connection::send_request(secs2::Message msg, ReplyHandler cb) {
}
void Connection::send_data(secs2::Message msg) {
send_data(device_id_, std::move(msg));
}
void Connection::send_data(uint16_t device_id, secs2::Message msg) {
if (closed_) return;
const uint32_t sys = next_system_bytes();
Frame f(Header::data_message(device_id_, msg.stream, msg.function, msg.reply_expected, sys),
Frame f(Header::data_message(device_id, msg.stream, msg.function, msg.reply_expected, sys),
msg.encode_body());
log("-> " + f.header.describe());
send_frame(std::move(f));
@@ -436,12 +516,23 @@ void Connection::close(const std::string& reason) {
// --- handshakes & timers --------------------------------------------------
void Connection::send_select_req() {
// SS legacy: Select.req uses 0xFFFF session id. Used unchanged when
// running with a single registered session.
const uint32_t sys = next_system_bytes();
log("-> Select.req");
send_frame(Frame(Header::control(SType::SelectReq, sys)));
start_control_transaction(SType::SelectRsp, sys, "Select");
}
void Connection::send_select_req(uint16_t device_id) {
// GS: Select.req carries the session's device_id in the session_id
// field (E37 §11.4.2).
const uint32_t sys = next_system_bytes();
log("-> Select.req session=" + std::to_string(device_id));
send_frame(Frame(Header::control(SType::SelectReq, sys, device_id)));
start_control_transaction(SType::SelectRsp, sys, "Select");
}
void Connection::send_linktest_req() {
const uint32_t sys = next_system_bytes();
send_frame(Frame(Header::control(SType::LinktestReq, sys)));
@@ -449,12 +540,18 @@ void Connection::send_linktest_req() {
}
void Connection::enter_selected() {
if (state_ == State::Selected) return;
state_ = State::Selected;
enter_selected(device_id_);
}
void Connection::enter_selected(uint16_t device_id) {
auto* s = find_session(device_id);
if (!s) return;
if (s->state == State::Selected) return;
s->state = State::Selected;
t7_timer_.cancel();
log("== SELECTED ==");
log("== SELECTED session=" + std::to_string(device_id) + " ==");
arm_linktest();
if (on_selected_) on_selected_();
if (s->on_selected) s->on_selected();
}
void Connection::arm_t7() {
@@ -463,17 +560,30 @@ void Connection::arm_t7() {
t7_timer_.expires_after(timers_.t7);
auto self = shared_from_this();
t7_timer_.async_wait([this, self](std::error_code ec) {
if (!ec && !closed_ && state_ == State::NotSelected) close("T7 not-selected timeout");
if (!ec && !closed_) {
// Fire only if NO session has been selected — T7 is a single
// connection-scope timer per E37 §10, not per-session.
bool any = false;
for (auto& kv : sessions_) if (kv.second.state == State::Selected) { any = true; break; }
if (!any) close("T7 not-selected timeout");
}
});
}
void Connection::arm_linktest() {
if (timers_.linktest.count() <= 0) return;
if (state_ != State::Selected) return;
// Run linktest while any session is selected — that proves the
// underlying TCP path is alive regardless of which session uses it.
bool any = false;
for (auto& kv : sessions_) if (kv.second.state == State::Selected) { any = true; break; }
if (!any) return;
linktest_timer_.expires_after(timers_.linktest);
auto self = shared_from_this();
linktest_timer_.async_wait([this, self](std::error_code ec) {
if (!ec && !closed_ && state_ == State::Selected) send_linktest_req();
if (ec || closed_) return;
bool any2 = false;
for (auto& kv : sessions_) if (kv.second.state == State::Selected) { any2 = true; break; }
if (any2) send_linktest_req();
});
}