Skip to content

Allow IS-07 and IS-12 WebSockets to share a port #525

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

Open
iibaranov-IG wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Allow IS-07 and IS-12 WebSockets to share a port #525

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6dd1731
Allow IS-07 and IS-12 WebSockets to share a port
iibaranov-IG Oct 1, 2026
62ff171
Remove control_protocol_websockets from the shared common server code…
lo-simon Oct 26, 2026
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
1 change: 1 addition & 0 deletions Development/cmake/NmosCppTest.cmake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ set(NMOS_CPP_TEST_NMOS_TEST_SOURCES
nmos/test/log_gate_test.cpp
nmos/test/mdns_test.cpp
nmos/test/node_interfaces_test.cpp
nmos/test/node_server_test.cpp
nmos/test/paging_utils_test.cpp
nmos/test/query_api_test.cpp
nmos/test/resources_test.cpp
Expand Down
27 changes: 27 additions & 0 deletions Development/nmos/api_utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
#include "nmos/api_utils.h"

#include <cstddef>
#include <boost/algorithm/cxx11/any_of.hpp>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/algorithm/string/trim.hpp>
Expand Down Expand Up @@ -764,6 +765,32 @@ namespace nmos
return ws_listener;
}

web::websockets::experimental::listener::websocket_listener_handlers make_ws_api_router(std::map<utility::string_t, web::websockets::experimental::listener::websocket_listener_handlers> handlers)
{
const auto find_handlers = [handlers](const web::uri& uri) -> const web::websockets::experimental::listener::websocket_listener_handlers*
{
const auto& path = uri.path();
const web::websockets::experimental::listener::websocket_listener_handlers* result{};
std::size_t longest_path{};
for (const auto& handler : handlers)
{
if (boost::starts_with(path, handler.first) && longest_path < handler.first.size())
{
result = &handler.second;
longest_path = handler.first.size();
}
}
return result;
};

return {
[find_handlers](web::http::http_request req) { const auto handlers = find_handlers(req.request_uri()); return handlers && handlers->validate_handler(req); },
[find_handlers](const web::uri& uri, const web::websockets::experimental::listener::connection_id& id) { if (const auto handlers = find_handlers(uri)) handlers->open_handler(uri, id); },
[find_handlers](const web::uri& uri, const web::websockets::experimental::listener::connection_id& id, web::websockets::websocket_close_status status, const utility::string_t& reason) { if (const auto handlers = find_handlers(uri)) handlers->close_handler(uri, id, status, reason); },
[find_handlers](const web::uri& uri, const web::websockets::experimental::listener::connection_id& id, const web::websockets::websocket_incoming_message& message) { if (const auto handlers = find_handlers(uri)) handlers->message_handler(uri, id, message); }
Comment on lines +787 to +790

@garethsb garethsb Nov 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

would it make sense to capture find_handlers via a shared_ptr to avoid making four copies? not sure

};
}

// returns "http" or "https" depending on settings
utility::string_t http_scheme(const nmos::settings& settings)
{
Expand Down
3 changes: 3 additions & 0 deletions Development/nmos/api_utils.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,9 @@ namespace nmos
// construct a websocket_listener on the specified address and port - captures handlers by reference!
web::websockets::experimental::listener::websocket_listener make_ws_api_listener(bool secure, const utility::string_t& host_address, int port, const web::websockets::experimental::listener::websocket_listener_handlers& handlers, web::websockets::experimental::listener::websocket_listener_config config, slog::base_gate& gate);

// combine WebSocket API handlers for one listener, selecting an API by the leading path component
web::websockets::experimental::listener::websocket_listener_handlers make_ws_api_router(std::map<utility::string_t, web::websockets::experimental::listener::websocket_listener_handlers> handlers);

// returns "http" or "https" depending on settings
utility::string_t http_scheme(const nmos::settings& settings);

Expand Down
9 changes: 6 additions & 3 deletions Development/nmos/control_protocol_ws_api.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
#include "nmos/control_protocol_ws_api.h"

#include <algorithm>
#include <boost/algorithm/string/join.hpp>
#include <boost/range/join.hpp>
#include "cpprest/json_validator.h"
Expand Down Expand Up @@ -65,11 +66,13 @@ namespace nmos
if (!ws_validate_authorization(req, nmos::experimental::scopes::ncp)) { return false; }
}

// For now just return true
const auto& ws_ncp_path = req.request_uri().path();
slog::log<slog::severities::more_info>(gate, SLOG_FLF) << "Validating websocket connection to: " << ws_ncp_path;

return true;
const auto is12_versions = nmos::is12_versions::from_settings(model.settings);

@garethsb garethsb Nov 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Needs at least a read lock (and might as well use the same simple versions name), like:

nmos-cpp/Development/nmos/node_api.cpp

Line 41 in b6c8367

const auto versions = with_read_lock(model.mutex, [&model] { return nmos::is04_versions::from_settings(model.settings); });

return std::any_of(is12_versions.begin(), is12_versions.end(), [&ws_ncp_path](const nmos::api_version& version)
{
return ws_ncp_path == U("/x-nmos/ncp/") + nmos::make_api_version(version);
});
};
}

Expand Down
59 changes: 29 additions & 30 deletions Development/nmos/node_server.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,17 +86,23 @@ namespace nmos
node_server.api_routers[{ {}, nmos::fields::configuration_port(node_model.settings) }].mount({}, nmos::make_configuration_api(node_model, validate_authorization ? validate_authorization(nmos::experimental::scopes::configuration) : nullptr, node_implementation.get_control_protocol_class_descriptor, node_implementation.get_control_protocol_datatype_descriptor, node_implementation.get_control_protocol_method_descriptor, node_implementation.create_validation_fingerprint, node_implementation.validate_validation_fingerprint, node_implementation.get_read_only_modification_allow_list, node_implementation.remove_device_model_object, node_implementation.create_device_model_object, node_implementation.control_protocol_property_changed, gate));

const auto& events_ws_port = nmos::fields::events_ws_port(node_model.settings);
auto& events_ws_api = node_server.ws_handlers[{ {}, nmos::fields::events_ws_port(node_model.settings) }];
events_ws_api.first = nmos::make_events_ws_api(node_model, events_ws_api.second, node_implementation.ws_validate_authorization, gate);

// can't share a port between the events ws and the control protocol ws
const auto& control_protocol_enabled = (0 <= nmos::fields::control_protocol_ws_port(node_model.settings));
const auto& control_protocol_ws_port = nmos::fields::control_protocol_ws_port(node_model.settings);

static const utility::string_t events_ws_path_prefix{ U("/x-nmos/events/") };
static const utility::string_t control_protocol_ws_path_prefix{ U("/x-nmos/ncp/") };

// events ws handlers
{
auto& events_ws_entry = node_server.ws_handlers[{ {}, events_ws_port }][events_ws_path_prefix];
events_ws_entry.first = nmos::make_events_ws_api(node_model, events_ws_entry.second, node_implementation.ws_validate_authorization, gate);
}

// control protocol ws handlers
if (control_protocol_enabled)
{
if (control_protocol_ws_port == events_ws_port) throw std::runtime_error("Same port used for events and control protocol websockets are not supported");
auto& control_protocol_ws_api = node_server.ws_handlers[{ {}, control_protocol_ws_port }];
control_protocol_ws_api.first = nmos::make_control_protocol_ws_api(node_model, control_protocol_ws_api.second, node_implementation.ws_validate_authorization, node_implementation.get_control_protocol_class_descriptor, node_implementation.get_control_protocol_datatype_descriptor, node_implementation.get_control_protocol_method_descriptor, node_implementation.control_protocol_property_changed, gate);
auto& control_protocol_ws_entry = node_server.ws_handlers[{ {}, control_protocol_ws_port }][control_protocol_ws_path_prefix];
control_protocol_ws_entry.first = nmos::make_control_protocol_ws_api(node_model, control_protocol_ws_entry.second, node_implementation.ws_validate_authorization, node_implementation.get_control_protocol_class_descriptor, node_implementation.get_control_protocol_datatype_descriptor, node_implementation.get_control_protocol_method_descriptor, node_implementation.control_protocol_property_changed, gate);
}

// Set up the listeners for each HTTP API port
Expand All @@ -117,32 +123,25 @@ namespace nmos
auto websocket_config = nmos::make_websocket_listener_config(node_model.settings, node_implementation.load_server_certificates, node_implementation.load_dh_param, node_implementation.get_ocsp_response, gate);
websocket_config.set_log_callback(nmos::make_slog_logging_callback(gate));

size_t event_ws_pos{ 0 };
bool found_event_ws{ false };
size_t control_protocol_ws_pos{ 0 };
bool found_control_protocol_ws{ false };
for (auto& ws_handler : node_server.ws_handlers)
{
// combine the handlers for each API sharing this WebSocket port into one set of handlers
auto& api_handlers = ws_handler.second;
std::map<utility::string_t, web::websockets::experimental::listener::websocket_listener_handlers> path_handlers;
for (auto& api_handler : api_handlers)
{
path_handlers[api_handler.first] = api_handler.second.first;
}
node_server.combined_ws_handlers.push_back(nmos::make_ws_api_router(path_handlers));

// if IP address isn't specified for this router, use default server address or wildcard address
const auto& host = !ws_handler.first.first.empty() ? ws_handler.first.first : !server_address.empty() ? server_address : web::websockets::experimental::listener::host_wildcard;
// map the configured client port to the server port on which to listen
// hmm, this should probably also take account of the address
node_server.ws_listeners.push_back(nmos::make_ws_api_listener(server_secure, host, nmos::experimental::server_port(ws_handler.first.second, node_model.settings), ws_handler.second.first, websocket_config, gate));

if (!found_event_ws)
{
if (ws_handler.first.second == events_ws_port) { found_event_ws = true; }
else { ++event_ws_pos; }
}

if (control_protocol_enabled && !found_control_protocol_ws)
{
if (ws_handler.first.second == control_protocol_ws_port) { found_control_protocol_ws = true; }
else { ++control_protocol_ws_pos; }
}
node_server.ws_listeners.insert({ ws_handler.first, nmos::make_ws_api_listener(server_secure, host, nmos::experimental::server_port(ws_handler.first.second, node_model.settings), node_server.combined_ws_handlers.back(), websocket_config, gate) });
}

auto& events_ws_listener = node_server.ws_listeners.at(event_ws_pos);
auto& events_ws_listener = node_server.ws_listeners.at({ {}, events_ws_port });

// Set up node operation (including the DNS-SD advertisements)

Expand All @@ -154,9 +153,10 @@ namespace nmos
auto monitor_connection_activated = node_implementation.monitor_connection_activated;
auto channelmapping_activated = node_implementation.channelmapping_activated;
auto get_authorization_bearer_token = node_implementation.get_authorization_bearer_token;
auto& event_ws = node_server.ws_handlers.at({ {}, events_ws_port }).at(events_ws_path_prefix).second;
node_server.thread_functions.assign({
[&, load_ca_certificates, registration_changed, get_authorization_bearer_token] { nmos::node_behaviour_thread(node_model, load_ca_certificates, registration_changed, get_authorization_bearer_token, gate); },
[&] { nmos::send_events_ws_messages_thread(events_ws_listener, node_model, events_ws_api.second, gate); },
[&] { nmos::send_events_ws_messages_thread(events_ws_listener, node_model, event_ws, gate); },
[&] { nmos::erase_expired_events_resources_thread(node_model, gate); },
[&, resolve_auto, set_transportfile, connection_activated, monitor_connection_activated] { nmos::connection_activation_thread(node_model, resolve_auto, set_transportfile, connection_activated, monitor_connection_activated, gate); },
[&, channelmapping_activated] { nmos::channelmapping_activation_thread(node_model, channelmapping_activated, gate); }
Expand All @@ -167,12 +167,11 @@ namespace nmos
{
node_server.thread_functions.push_back([&, load_ca_certificates, system_changed] { nmos::node_system_behaviour_thread(node_model, load_ca_certificates, system_changed, gate); });
}

if (control_protocol_enabled)
{
auto& control_protocol_ws_listener = node_server.ws_listeners.at(control_protocol_ws_pos);
auto& control_protocol_ws_api = node_server.ws_handlers.at({ {}, control_protocol_ws_port });
node_server.thread_functions.push_back([&] { nmos::send_control_protocol_ws_messages_thread(control_protocol_ws_listener, node_model, control_protocol_ws_api.second, gate); });
auto& control_protocol_ws_listener = node_server.ws_listeners.at({ {}, control_protocol_ws_port });
auto& control_protocol_ws = node_server.ws_handlers.at({ {}, control_protocol_ws_port }).at(control_protocol_ws_path_prefix).second;
node_server.thread_functions.push_back([&] { nmos::send_control_protocol_ws_messages_thread(control_protocol_ws_listener, node_model, control_protocol_ws, gate); });
}

return node_server;
Expand Down
25 changes: 20 additions & 5 deletions Development/nmos/registry_server.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,8 +89,13 @@ namespace nmos
// See https://specs.amwa.tv/is-04/releases/v1.2.0/APIs/schemas/with-refs/queryapi-subscriptions-websocket.html
const nmos::id query_id = nmos::make_repeatable_id(nmos::experimental::fields::seed_id(registry_model.settings), U("/x-nmos/query"));

auto& query_ws_api = registry_server.ws_handlers[{ {}, nmos::fields::query_ws_port(registry_model.settings) }];
query_ws_api.first = nmos::make_query_ws_api(query_id, registry_model, query_ws_api.second, registry_implementation.ws_validate_authorization, gate);
static const utility::string_t query_ws_path_prefix{ U("/x-nmos/query/") };

// query ws handlers
{
auto& query_ws_entry = registry_server.ws_handlers[{ {}, nmos::fields::query_ws_port(registry_model.settings) }][query_ws_path_prefix];
query_ws_entry.first = nmos::make_query_ws_api(query_id, registry_model, query_ws_entry.second, registry_implementation.ws_validate_authorization, gate);
}

// Configure the Registration API

Expand Down Expand Up @@ -146,19 +151,29 @@ namespace nmos

for (auto& ws_handler : registry_server.ws_handlers)
{
// combine the handlers for each API sharing this WebSocket port into one set of handlers
auto& api_handlers = ws_handler.second;
std::map<utility::string_t, web::websockets::experimental::listener::websocket_listener_handlers> path_handlers;
for (auto& api_handler : api_handlers)
{
path_handlers[api_handler.first] = api_handler.second.first;
}
registry_server.combined_ws_handlers.push_back(nmos::make_ws_api_router(path_handlers));

// if IP address isn't specified for this router, use default server address or wildcard address
const auto& host = !ws_handler.first.first.empty() ? ws_handler.first.first : !server_address.empty() ? server_address : web::websockets::experimental::listener::host_wildcard;
// map the configured client port to the server port on which to listen
// hmm, this should probably also take account of the address
registry_server.ws_listeners.push_back(nmos::make_ws_api_listener(server_secure, host, nmos::experimental::server_port(ws_handler.first.second, registry_model.settings), ws_handler.second.first, websocket_config, gate));
registry_server.ws_listeners.insert({ ws_handler.first, nmos::make_ws_api_listener(server_secure, host, nmos::experimental::server_port(ws_handler.first.second, registry_model.settings), registry_server.combined_ws_handlers.back(), websocket_config, gate) });
}

auto& query_ws_listener = registry_server.ws_listeners.back();
auto& query_ws_listener = registry_server.ws_listeners.at({ {}, nmos::fields::query_ws_port(registry_model.settings) });

// Set up registry management (including the DNS-SD advertisements)

auto& query_ws = registry_server.ws_handlers.at({ {}, nmos::fields::query_ws_port(registry_model.settings) }).at(query_ws_path_prefix).second;
registry_server.thread_functions.assign({
[&] { nmos::send_query_ws_events_thread(query_ws_listener, registry_model, query_ws_api.second, gate); },
[&] { nmos::send_query_ws_events_thread(query_ws_listener, registry_model, query_ws, gate); },
[&] { nmos::erase_expired_resources_thread(registry_model, gate); },
[&] { nmos::advertise_registry_thread(registry_model, gate); }
});
Expand Down
4 changes: 2 additions & 2 deletions Development/nmos/server.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ namespace nmos

for (auto& ws_listener : ws_listeners)
{
if (0 <= ws_listener.uri().port()) tasks.push_back(ws_listener.open());
if (0 <= ws_listener.second.uri().port()) tasks.push_back(ws_listener.second.open());
}

return pplx::ranges::when_all(tasks).then(pplx::observe_exceptions(tasks));
Expand All @@ -83,7 +83,7 @@ namespace nmos
}
for (auto& ws_listener : ws_listeners)
{
if (0 <= ws_listener.uri().port()) tasks.push_back(ws_listener.close());
if (0 <= ws_listener.second.uri().port()) tasks.push_back(ws_listener.second.close());
}

return pplx::ranges::when_all(tasks).then(pplx::observe_exceptions(tasks));
Expand Down
8 changes: 6 additions & 2 deletions Development/nmos/server.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,12 @@ namespace nmos
std::vector<web::http::experimental::listener::http_listener> http_listeners;

// WebSocket APIs
std::map<host_port, std::pair<web::websockets::experimental::listener::websocket_listener_handlers, nmos::websockets>> ws_handlers;
std::vector<web::websockets::experimental::listener::websocket_listener> ws_listeners;
// each port may be shared by multiple APIs, each identified by a path prefix (e.g. "/x-nmos/events/")
// in order to maintain a separate set of connections (nmos::websockets) per API, even when a port is shared
typedef std::map<utility::string_t, std::pair<web::websockets::experimental::listener::websocket_listener_handlers, nmos::websockets>> ws_api_handlers;
std::map<host_port, ws_api_handlers> ws_handlers;
std::deque<web::websockets::experimental::listener::websocket_listener_handlers> combined_ws_handlers; // for APIs sharing a port, the handlers are combined into one set of handlers

@garethsb garethsb Oct 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why deque?

@lo-simon lo-simon Oct 26, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is because make_ws_api_listener() captures the handlers by reference; each websocket_listener ends up storing a std::reference_wrapper pointing directly at the memory address of a specific websocket_listener_handlers object living inside combined_ws_handlers.

If combined_ws_handlers were a std::vector, calling .push_back() in a loop (once per WebSocket port being configured) can trigger reallocation when the vector's capacity is exceeded. Reallocation copies/moves all existing elements to a new, larger memory block and frees the old one. That invalidates every reference/pointer/iterator to previously inserted elements; see example below:

e.g. if with a std::vector

for (auto& ws_handler : node_server.ws_handlers)   // e.g. 2 iterations: events port, control protocol port
{
    ...
    node_server.combined_ws_handlers.push_back(nmos::make_ws_api_router(path_handlers));
    // if this push_back reallocates, any websocket_listener already
    // constructed in a PREVIOUS iteration (which holds std::ref to an
    // earlier combined_ws_handlers element) is now dangling!
    node_server.ws_listeners.insert({ ws_handler.first, nmos::make_ws_api_listener(..., node_server.combined_ws_handlers.back(), ...) });
}

But std::deque guarantees that inserting at the front or back never invalidates references or pointers to other elements (only iterators may be invalidated, but we're not holding onto iterators - we're holding onto references inside std::ref, which deque preserves across push_back). Internally, deque is implemented as a sequence of fixed-size chunks, so growing it just adds a new chunk rather than reallocating/moving all existing elements.

This matches exactly what we need here: every websocket_listener_handlers added in earlier loop iterations must remain at a stable address for the entire lifetime of node_server, even as more are added in later iterations.

@garethsb garethsb Nov 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes sense, thanks!

std::map<host_port, websocket_listener_handlers> would keep references stable too, and would be keyed the same way as ws_listeners, so each listener's handlers are easy to find. Or we could keep each listener and its handlers together as a pair value in one map? Or maybe no need.

std::map<host_port, web::websockets::experimental::listener::websocket_listener> ws_listeners;

// Server behaviours
std::vector<std::function<void()>> thread_functions;
Expand Down
Loading
Loading