diff --git a/.gitattributes b/.gitattributes new file mode 100644 index 00000000..7c32d5f7 --- /dev/null +++ b/.gitattributes @@ -0,0 +1 @@ +*.jar filter=lfs diff=lfs merge=lfs -text diff --git a/.restyled.yaml b/.restyled.yaml index 95df9d9c..cb8d816e 100644 --- a/.restyled.yaml +++ b/.restyled.yaml @@ -19,6 +19,9 @@ ignore_labels: restylers_version: stable +also_exclude: + - "**/*.jar" + restylers: - clang-format: image: diff --git a/CMakeLists.txt b/CMakeLists.txt index 489b2c8b..0aaba829 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -13,6 +13,12 @@ if(NOT DEFINED GR_DIGITIZERS_TOPLEVEL_PROJECT) endif() endif() +if(NOT EMSCRIPTEN) + option(ENABLE_PICOSCOPE "Enable PicoScope support" ON) + option(ENABLE_TIMING "Enable TimingReceiver support" ON) + option(ENABLE_OPENCMW "Enable OpenCmw support" ON) +endif() + if(CMAKE_CXX_COMPILER_ID MATCHES ".*Clang") # set default C++ STL to Clang's # libc++ when using Clang add_compile_options(-stdlib=libc++ -fcolor-diagnostics) @@ -32,12 +38,25 @@ FetchContent_Declare( GIT_REPOSITORY https://github.com/fair-acc/gnuradio4.git GIT_TAG d14bb560909aaa151b13a535ad4c7f98ad3e9959 # main as of 2026-02-24 EXCLUDE_FROM_ALL) +list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST gnuradio4) + +if(ENABLE_GR_DIGITIZERS_TESTING) + FetchContent_Declare( + ut + GIT_REPOSITORY https://github.com/boost-ext/ut.git + GIT_TAG v2.3.1 # latest tag as of 2025-04-02 + EXCLUDE_FROM_ALL) + list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST ut) +endif() -FetchContent_Declare( - ut - GIT_REPOSITORY https://github.com/boost-ext/ut.git - GIT_TAG v2.3.1 # latest tag as of 2025-04-02 - EXCLUDE_FROM_ALL) +if(ENABLE_OPENCMW) + FetchContent_Declare( + opencmw-cpp + GIT_REPOSITORY https://github.com/fair-acc/opencmw-cpp.git + GIT_TAG e30b78193052bf2ea39160a8ec913044552fe103 # latest as of 2026-03-19 + EXCLUDE_FROM_ALL) + list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST "opencmw-cpp") +endif() if(CMAKE_CXX_COMPILER_ID MATCHES "(Clang|GNU)") if(ADDRESS_SANITIZER) @@ -68,7 +87,7 @@ include(cmake/CompilerWarnings.cmake) set_project_warnings(gr-digitizers-options) if(EMSCRIPTEN) - FetchContent_MakeAvailable(gnuradio4 ut) + FetchContent_MakeAvailable(${GR_DIGITIZER_FETCH_CONTENT_LIST}) set(CMAKE_EXECUTABLE_SUFFIX ".js") target_compile_options(gr-digitizers-options INTERFACE -fwasm-exceptions -pthread) @@ -87,6 +106,7 @@ else() imgui GIT_REPOSITORY https://github.com/ocornut/imgui.git GIT_TAG v1.91.9b-docking) + list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST imgui) # Enables 32 bit vertex indices for ImGui add_compile_definitions("ImDrawIdx=unsigned int") @@ -94,7 +114,8 @@ else() implot GIT_REPOSITORY https://github.com/epezent/implot.git GIT_TAG v0.16) - FetchContent_MakeAvailable(imgui implot gnuradio4 ut) + list(APPEND GR_DIGITIZER_FETCH_CONTENT_LIST implot) + FetchContent_MakeAvailable(${GR_DIGITIZER_FETCH_CONTENT_LIST}) find_package(SDL3 REQUIRED) find_package(OpenGL REQUIRED COMPONENTS OpenGL) @@ -137,11 +158,6 @@ if(CMAKE_CXX_COMPILER MATCHES "/em\\+\\+(-[a-zA-Z0-9.])?$") # if this hasn't set(EMSCRIPTEN true) endif() -if(NOT EMSCRIPTEN) - option(ENABLE_PICOSCOPE "Enable PicoScope support" ON) - option(ENABLE_TIMING "Enable TimingReceiver support" ON) -endif() - find_package(PkgConfig REQUIRED) if(ENABLE_PICOSCOPE) diff --git a/blocklib/CMakeLists.txt b/blocklib/CMakeLists.txt index 8e7ec5e3..74879827 100644 --- a/blocklib/CMakeLists.txt +++ b/blocklib/CMakeLists.txt @@ -5,3 +5,7 @@ endif() if(ENABLE_TIMING) add_subdirectory(timing) endif() + +if(ENABLE_OPENCMW) + add_subdirectory(opencmw) +endif() diff --git a/blocklib/opencmw/CMakeLists.txt b/blocklib/opencmw/CMakeLists.txt new file mode 100644 index 00000000..e8406746 --- /dev/null +++ b/blocklib/opencmw/CMakeLists.txt @@ -0,0 +1,16 @@ +add_library(fair-opencmw INTERFACE) +target_include_directories( + fair-opencmw INTERFACE $ + $) + +target_link_libraries(fair-opencmw INTERFACE gr-digitizers-options client) +set_target_properties( + fair-opencmw + PROPERTIES + PUBLIC_HEADER + "include/fair/opencmw/cmwlight/DirectoryLightClient.hpp;include/fair/opencmw/cmwlight/CmwLightClient.hpp;include/fair/opencmw/OpenCmwSource.hpp" +) + +if(ENABLE_GR_DIGITIZERS_TESTING) + add_subdirectory(test) +endif() diff --git a/blocklib/opencmw/include/fair/opencmw/OpenCmwSource.hpp b/blocklib/opencmw/include/fair/opencmw/OpenCmwSource.hpp new file mode 100644 index 00000000..00acbd80 --- /dev/null +++ b/blocklib/opencmw/include/fair/opencmw/OpenCmwSource.hpp @@ -0,0 +1,164 @@ +#ifndef GR_DIGITIZERS_OPENCMWSOURCE_HPP +#define GR_DIGITIZERS_OPENCMWSOURCE_HPP + +#include +#include +#include +#include +#include +#include +#include + +namespace fair::opencmw { +static ::opencmw::client::ClientContext initGlobalClientContext(::opencmw::zmq::Context& zctx) { + using namespace ::opencmw::client; + std::string nameserver = "http://localhost:7500"; + if (const auto ptr = std::getenv("CMW_NAMESERVER"); ptr != nullptr) { + nameserver = std::string{ptr}; + } + std::vector> clients{}; + clients.emplace_back(std::make_unique()); + clients.emplace_back(std::make_unique(zctx, 100ms, "OpenCmwMajordomoClient")); + clients.emplace_back(std::make_unique(zctx, nameserver, 100ms, "OpenCmwRda3Client")); + return ClientContext{std::move(clients)}; +} + +static ::opencmw::client::ClientContext& getGloablClientContext() { + static ::opencmw::zmq::Context zctx{}; + static ::opencmw::client::ClientContext clientContext = initGlobalClientContext(zctx); + return clientContext; +} + +struct OpenCmwSource : gr::Block { + using FloatLimits = gr::Limits::lowest(), std::numeric_limits::max()>; + gr::PortOut out; + + gr::Annotated> url; + gr::Annotated> signal_name; + gr::Annotated> signal_quantity; + gr::Annotated> signal_unit; + gr::Annotated, FloatLimits> signal_min = std::numeric_limits::lowest(); + gr::Annotated, FloatLimits> signal_max = std::numeric_limits::max(); + gr::Annotated> verbose_console = false; + gr::Annotated> reconnect_timeout = 5.f; + gr::Annotated> nameserver = "http://localhost:7500"; + + GR_MAKE_REFLECTABLE(OpenCmwSource, out, url, signal_name, signal_unit, signal_quantity, signal_min, signal_max, verbose_console, reconnect_timeout); + + std::mutex mutex; + std::queue updates; + + void start() { + if (verbose_console) { + std::println("OpenCmwSource: subscribe to {}", url); + } + auto& clientContext = getGloablClientContext(); + clientContext.subscribe(::opencmw::URI(url), [this](const ::opencmw::mdp::Message& response) { + if (verbose_console) { + std::println("OpenCmwSource: received update"); + } + auto res = gr::property_map{}; + auto header = gr::property_map{}; + header.try_emplace("url", std::pmr::string{url}); + if (response.error.empty()) { + using AcquisitionMap = std::map, std::vector, std::vector, std::vector, std::vector, std::vector, std::vector, std::vector>>; + ::opencmw::IoBuffer buffer(response.data); + AcquisitionMap acqMap{}; + ::opencmw::DeserialiserInfo info; + ::opencmw::FieldDescriptionLong field; + ::opencmw::FieldHeaderReader<::opencmw::CmwLight>::get<::opencmw::ProtocolCheck::LENIENT>(buffer, info, field); + ::opencmw::IoSerialiser<::opencmw::CmwLight, AcquisitionMap>::deserialise(buffer, field, acqMap); + if (verbose_console) { + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}\n errors: {}\n mapFields: {}", info.additionalFields | std::views::keys, info.additionalFields | std::views::values, info.exceptions | std::views::transform([](auto& e) { return e.what(); }), acqMap | std::views::keys); + } + auto val = gr::property_map{}; + for (auto& [key, value] : acqMap) { + std::visit([&](auto&& v) { val.insert(std::make_pair(key, gr::pmt::Value(v))); }, value); + } + res.try_emplace("data", val); + auto errors = gr::pmt::Value{info.exceptions | std::views::transform([](auto& e) { return e.what(); }) | std::ranges::to>()}; + res.try_emplace("errors", errors); + } else { + if (verbose_console) { + std::println("error in subscription: {}", response.error); + } + auto errors = gr::pmt::Value{response.error}; + res.try_emplace("errors", errors); + } + res.try_emplace("header", header); + std::lock_guard guard{mutex}; + updates.push(std::move(res)); + this->progress->incrementAndGet(); + }); + } + + void stop() { + auto& clientContext = getGloablClientContext(); + if (verbose_console) { + std::println("OpenCmwSource: unsubscribe from {}", url); + } + clientContext.unsubscribe(::opencmw::URI(url)); + } + + auto processBulk(gr::OutputSpanLike auto& output) noexcept { + size_t written = 0; + std::lock_guard guard{mutex}; + while (written < output.size() && !updates.empty()) { + if (verbose_console) { + std::println("OpenCmwSource: publish update(s): {}", updates.front()); + } + *(output.begin() + written) = updates.front(); + ++written; + updates.pop(); + } + output.publish(written); + // todo: propagate metadata + return written == 0UZ ? gr::work::Status::INSUFFICIENT_INPUT_ITEMS : gr::work::Status::OK; + } +}; + +template +struct ExtractFromMap : gr::Block> { + gr::Annotated> fieldname; + + gr::PortIn in; + gr::PortOut out; + + GR_MAKE_REFLECTABLE(ExtractFromMap, in, out, fieldname); + + [[nodiscard]] constexpr gr::work::Status processBulk(gr::InputSpanLike auto& input, gr::OutputSpanLike auto& output) noexcept { + std::size_t nIn = 0; + std::size_t nOut = 0; + for (const auto& map : input) { + if (const gr::pmt::Value::Map* dataPtr = map.template get_if(); dataPtr != nullptr && dataPtr->contains("data")) { + if (const auto* mapPtr = dataPtr->at("data").get_if(); mapPtr != nullptr && mapPtr->contains(fieldname)) { + if (const T* valuePtr = mapPtr->at(std::pmr::string{fieldname.value}).get_if(); valuePtr != nullptr) { + if (nOut >= output.size()) { + break; + } + output[nOut] = *valuePtr; + ++nOut; + } else if (const gr::Tensor* tensorPtr = mapPtr->at(std::pmr::string{fieldname.value}).get_if>(); tensorPtr != nullptr) { + if (nOut + tensorPtr->size() >= output.size()) { + break; + } + std::copy(tensorPtr->begin(), tensorPtr->end(), output.begin() + nOut); + nOut += tensorPtr->size(); + } else { + std::println("field {1} exists, but has unsupported field type: {2}, supported: {0} or Tensor<{0}>", gr::meta::type_name(), fieldname.value, magic_enum::enum_name<>(magic_enum::enum_cast(mapPtr->at(std::pmr::string{fieldname.value})._value_type).value_or(gr::pmt::Value::ValueType::Monostate))); + } + } + } + ++nIn; + } + output.publish(nOut); + input.consume(nIn); + return nOut > 0 ? gr::work::Status::OK : (nIn > 0 ? gr::work::Status::INSUFFICIENT_OUTPUT_ITEMS : gr::work::Status::INSUFFICIENT_INPUT_ITEMS); + } +}; + +} // namespace fair::opencmw + +inline auto registerRemoteStreamSource = ::gr::registerBlock(::gr::globalBlockRegistry()); + +#endif // GR_DIGITIZERS_OPENCMWSOURCE_HPP diff --git a/blocklib/opencmw/include/fair/opencmw/cmwlight/CmwLightClient.hpp b/blocklib/opencmw/include/fair/opencmw/cmwlight/CmwLightClient.hpp new file mode 100644 index 00000000..425becb0 --- /dev/null +++ b/blocklib/opencmw/include/fair/opencmw/cmwlight/CmwLightClient.hpp @@ -0,0 +1,923 @@ +#ifndef OPENCMW_CPP_CMWLIGHTCLIENT_HPP +#define OPENCMW_CPP_CMWLIGHTCLIENT_HPP + +#include +#include +#include +#include + +#include "DirectoryLightClient.hpp" +#include +#include +#include +#include +#include +#include +#include + +namespace opencmw::client::cmwlight { + +struct Request { + URI<> uri; + std::function callback; + timePoint timestamp_received = std::chrono::system_clock::now(); +}; + +struct Subscription { + URI<> uri; + std::function callback; + timePoint timestamp_received = std::chrono::system_clock::now(); + std::size_t reqId; +}; + +struct CmwLightError { + using DataMap = std::map>; + int64_t ContextAcqStamp{}; + int64_t ContextCycleStamp{}; + std::string Message; + int8_t Type{}; + DataMap ContextData{}; +}; + +struct CmwLightHeaderOptions { + using StringMap = std::map; + int64_t b{}; // SOURCE_ID + StringMap e; // SESSION_BODY + // can potentially contain more and arbitrary data + // accessors to make code more readable + int64_t& sourceId() { return b; } + StringMap& sessionBody() { return e; } +}; + +struct CmwLightHeader { + int8_t x_2{}; // REQ_TYPE_TAG + int64_t x_0{}; // ID_TAG + std::string x_1; // DEVICE_NAME + std::string f; // PROPERTY_NAME + int8_t x_7{}; // UPDATE_TYPE + std::string d; // SESSION_ID + std::unique_ptr x_3; + // accessors to make code more readable + int8_t& requestType() { return x_2; } + int64_t& id() { return x_0; } + std::string& device() { return x_1; } + std::string& property() { return f; } + int8_t& updateType() { return x_7; } + std::string& sessionId() { return d; } + std::unique_ptr& options() { return x_3; } +}; + +struct CmwLightConnectBody { + std::string x_9; + std::string& clientInfo() { return x_9; } +}; + +struct CmwLightRequestContext { + using FilterMap = std::map>; + std::string x_8; // SELECTOR + FilterMap c; // FILTERS + FilterMap x; // DATA + // accessors to make code more readable + std::string& selector() { return x_8; }; + FilterMap& filters() { return c; } + FilterMap& data() { return x; } +}; + +struct CmwLightDataContext { + using StringMap = std::map; + std::string x_4; // CYCLE_NAME + int64_t x_6; // CYCLE_STAMP + int64_t x_5; // ACQ_STAMP + StringMap x; // DATA + // accessors to make code more readable + std::string& cycleName() { return x_4; } + long& cycleStamp() { return x_6; } + long& acqStamp() { return x_5; } + StringMap& data() { return x; } +}; + +} // namespace opencmw::client::cmwlight +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::CmwLightError, ContextAcqStamp, ContextCycleStamp, Message, Type, ContextData) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::CmwLightHeaderOptions, b, e) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::CmwLightHeader, x_2, x_0, x_1, f, x_7, d, x_3) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::CmwLightConnectBody, x_9) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::CmwLightRequestContext, x_8, c, x) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::CmwLightDataContext, x_4, x_6, x_5, x) + +namespace opencmw::client::cmwlight { +namespace detail { +/** + * Sent as the first frame of an RDA3 message determining the type of message + */ +enum class MessageType : char { SERVER_CONNECT_ACK = 0x01, SERVER_REP = 0x02, SERVER_HB = 0x03, CLIENT_CONNECT = 0x20, CLIENT_REQ = 0x21, CLIENT_HB = 0x22 }; + +/** + * Frame Types in the descriptor (Last frame of a message containing the type of each sub message) + */ +enum class FrameType : char { HEADER = 0, BODY = 1, BODY_DATA_CONTEXT = 2, BODY_REQUEST_CONTEXT = 3, BODY_EXCEPTION = 4 }; + +/** + * request type used in request header REQ_TYPE_TAG + */ +enum class RequestType : char { GET = 0, SET = 1, CONNECT = 2, REPLY = 3, EXCEPTION = 4, SUBSCRIBE = 5, UNSUBSCRIBE = 6, NOTIFICATION_DATA = 7, NOTIFICATION_EXC = 8, SUBSCRIBE_EXCEPTION = 9, EVENT = 10, SESSION_CONFIRM = 11 }; + +/** + * UpdateType + */ +enum class UpdateType : char { NORMAL = 0, FIRST_UPDATE = 1, IMMEDIATE_UPDATE = 2 }; + +inline std::string getHostName() { + std::string hostname; + hostname.resize(255); + if (const int result = gethostname(hostname.data(), hostname.capacity()); !result) { + hostname = "localhost"; + } else { + hostname.resize(strnlen(hostname.data(), hostname.size())); + hostname.shrink_to_fit(); + } + return hostname; +} + +inline std::string getIdentity() { + std::string hostname = getHostName(); + static int CONNECTION_ID_GENERATOR = 1; + static int channelIdGenerator = 1; + return std::format("{}/{}/{}/{}", hostname, getpid(), ++CONNECTION_ID_GENERATOR, ++channelIdGenerator); // N.B. this scheme is parsed/enforced by CMW +} + +inline std::string percentEncode(const std::string_view& str) { + std::string result; + result.reserve(str.size() * 3); // this is the upper bound if all characters have to be percent encoded, avoids reallocation + for (const char c : str) { + if (isalnum(c)) { + result += c; + } else { + result += std::format("%{:02X}", static_cast(c)); + } + } + return result; +} + +inline std::string createClientInfo() { + const std::string hostname = getHostName(); + std::string username; + username.reserve(32); + if (const int usernameResult = getlogin_r(username.data(), username.capacity()); usernameResult == 0) { + username.resize(strnlen(username.data(), username.size())); + } else { + username = "unknown"; + } + const std::string processName = program_invocation_short_name; + const std::string language = "cpp"; + const long startTime = std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + const int pid = getpid(); + const std::string version = "2026.2.1"; + std::size_t applicationIdSize = 26 + processName.size() + username.size() + hostname.size() + std::to_string(pid).size(); + const std::string applicationId = std::format("app={};ver={};uid={};host={};pid={};", percentEncode(processName), percentEncode(version), percentEncode(username), percentEncode(hostname), std::to_string(pid)); + return std::format("9#Address:#string#{}#tcp:%2F%2F{}:0#ApplicationId:#string#{}#{}#UserName:#string#{}#{}#ProcessName:#string#{}#{}#Language:#string#{}#{}#StartTime:#long#{}#Name:#string#{}#{}#Pid:#int#{}#Version:#string#{}#{}", hostname.size() + 8, percentEncode(hostname), applicationIdSize, applicationId, username.size(), percentEncode(username), processName.size(), percentEncode(processName), language.size(), percentEncode(language), startTime, processName.size(), percentEncode(processName), pid, version.size(), percentEncode(version)); +} + +inline std::string createClientId() { + std::string username; + username.reserve(32); + if (const int usernameResult = getlogin_r(username.data(), username.capacity()); usernameResult == 0) { + username.resize(strnlen(username.data(), username.size())); + } else { + username = "unknown"; + } + std::string hostname = getHostName(); + const std::string processName = program_invocation_short_name; + const int pid = getpid(); + const std::string version = "2026.2.1"; + const auto startTime = std::chrono::system_clock::now(); + return std::format("RemoteHostInfoImpl[name={}; userName={}; appId=[app={};ver={};uid={};host={};pid={};]; process={}; pid={}; address=tcp://{}:0; startTime={:%F %R%z}; connectionTime=About ago; version={}; language=CPP]1", processName, username, processName, version, username, hostname, pid, processName, pid, hostname, startTime, version); +} + +struct PendingRequest { + enum class RequestState { INITIALIZED, WAITING, FINISHED }; + std::string reqId; + IoBuffer data{}; + RequestType requestType{RequestType::GET}; + RequestState state{RequestState::INITIALIZED}; + std::string uri; +}; + +struct OpenSubscription { + enum class SubscriptionState { INITIALIZED, SUBSCRIBING, SUBSCRIBED, UNSUBSCRIBING, UNSUBSCRIBED }; + std::chrono::milliseconds backOff = 20ms; + long updateId{}; + long reqId = 0L; + long replyId{}; + SubscriptionState state = SubscriptionState::SUBSCRIBING; + std::chrono::system_clock::time_point nextTry; + std::string uri; +}; + +struct Connection { + using SubscriptionMap = std::map>; + using RequestMap = std::map>; + enum class ConnectionState { + DISCONNECTED, + NS_LOOKUP, + CONNECTING1, + CONNECTING2, + CONNECTED, + }; + std::string _deviceName; + std::string _authority; + zmq::Socket _socket; + ConnectionState _connectionState = ConnectionState::DISCONNECTED; + timePoint _nextReconnectAttemptTimeStamp = std::chrono::system_clock::now(); + timePoint _lastHeartbeatReceived = std::chrono::system_clock::now(); + timePoint _lastHeartBeatSent = std::chrono::system_clock::now(); + std::chrono::milliseconds _backoff = 20ms; // implements exponential back-off to get + std::vector _frames{}; // currently received frames; will be accumulated until the message is complete + SubscriptionMap _subscriptions; // all subscriptions requested for (un)subscribe + int64_t _subscriptionIdGenerator = 1; + int64_t _requestIdGenerator = 1; + RequestMap _pendingRequests; + + Connection(const zmq::Context& context, const std::string_view authority, const int zmq_dealer_type) : _authority{authority}, _socket{context, zmq_dealer_type} { zmq::initializeSocket(_socket).assertSuccess(); } +}; + +static void send(const zmq::Socket& socket, const int param, std::string_view errorMsg, auto&& data) { + if (zmq::MessageFrame connectFrame{FWD(data)}; !connectFrame.send(socket, param).isValid()) { + throw std::runtime_error(errorMsg.data()); + } +} + +static std::string descriptorToString(auto... descriptor) { + std::string result{}; + result.reserve(sizeof...(descriptor)); + ((result.push_back(std::to_underlying(descriptor))), ...); + return result; +} + +static IoBuffer serialiseCmwLight(auto& requestType) { + IoBuffer buffer{}; + opencmw::serialise(buffer, requestType); + buffer.reset(); + return buffer; +} + +inline void sendConnectRequest(Connection& con) { + using namespace std::string_view_literals; + detail::send(con._socket, ZMQ_SNDMORE, "error sending get frame"sv, std::to_underlying(MessageType::CLIENT_REQ)); + CmwLightHeader header; + header.requestType() = std::to_underlying(RequestType::CONNECT); + header.id() = con._requestIdGenerator++; + header.options() = std::make_unique(); + send(con._socket, ZMQ_SNDMORE, "failed to send message header"sv, serialiseCmwLight(header)); // send message header + CmwLightConnectBody connectBody; + connectBody.clientInfo() = createClientInfo(); + send(con._socket, ZMQ_SNDMORE, "failed to send message header"sv, serialiseCmwLight(connectBody)); // send message header + using enum FrameType; + send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER, BODY)); +} +} // namespace detail + +class CMWLightClientBase { +public: + virtual ~CMWLightClientBase() = default; + virtual bool receive(mdp::Message& message) = 0; + virtual timePoint housekeeping(const timePoint& now) = 0; + virtual void get(const URI<>&, std::string_view) = 0; + virtual void set(const URI<>&, std::string_view, const std::span&) = 0; + virtual void subscribe(const URI<>&, std::string_view) = 0; + virtual void unsubscribe(const URI<>&, std::string_view) = 0; +}; + +class CMWLightClient : public CMWLightClientBase { + using timeUnit = std::chrono::milliseconds; + const timeUnit _clientTimeout; + const zmq::Context& _context; + DirectoryLightClient& _directoryClient; + const std::string _clientId; + const std::string _sourceName; + std::vector _connections; + std::vector& _pollItems; + constexpr static auto HEARTBEAT_INTERVAL = 2000ms; + constexpr static auto MAX_BACKOFF = 600s; + +public: + explicit CMWLightClient(const zmq::Context& context, std::vector& pollItems, DirectoryLightClient& directoryClient, const timeUnit timeout = 1s, std::string clientId = "") : _clientTimeout(timeout), _context(context), _directoryClient(directoryClient), _clientId(std::move(clientId)), _sourceName(std::format("CMWLightClient(clientId: {})", _clientId)), _pollItems(pollItems) {} + + void connect(detail::Connection& con) const { + using enum detail::Connection::ConnectionState; + using namespace std::string_view_literals; + if (std::string id = detail::getIdentity(); !zmq::invoke(zmq_setsockopt, con._socket, ZMQ_IDENTITY, id.data(), id.size()).isValid()) { // hostname/process/id/channel -- the server verifies this + throw std::runtime_error(std::format("failed set socket identity: {}", id)); + } + if (auto endpoint = std::format("tcp://{}", con._authority); zmq::invoke(zmq_connect, con._socket, endpoint).isValid()) { + _pollItems.push_back({.socket = con._socket.zmq_ptr, .fd = 0, .events = ZMQ_POLLIN, .revents = 0}); + } else { + throw std::runtime_error(std::format("failed connect to server: {}", endpoint)); + } + // send rda3 connect message + detail::send(con._socket, ZMQ_SNDMORE, "error sending connect frame"sv, std::to_underlying(detail::MessageType::CLIENT_CONNECT)); + detail::send(con._socket, 0, "error sending connect frame"sv, "1.0.0"); + con._connectionState = CONNECTING1; + } + + detail::Connection& findConnection(const URI<>& uri) { + const std::string uriPath = uri.path().value(); + const auto deviceEndPos = uriPath.find('/', 1); + const std::string_view deviceName = std::string_view{uriPath}.substr(1, (deviceEndPos == std::string_view::npos ? uriPath.size() : deviceEndPos) - 1); + const auto con = std::ranges::find_if(_connections, [deviceName](const detail::Connection& c) { return c._deviceName == deviceName; }); + if (con == _connections.end()) { + auto newCon = detail::Connection(_context, uri.authority().value_or(""), ZMQ_DEALER); + newCon._deviceName = deviceName; + _connections.push_back(std::move(newCon)); + return _connections.back(); + } + return *con; + } + + void get(const URI<>& uri, const std::string_view req_id) override { + using namespace std::string_view_literals; + using enum detail::FrameType; + auto& con = findConnection(uri); // send message header + detail::PendingRequest req{}; + req.reqId = req_id; + req.requestType = detail::RequestType::GET; + req.state = detail::PendingRequest::RequestState::INITIALIZED; + req.uri = uri.str(); + con._pendingRequests.try_emplace(std::string(req_id), std::move(req)); + } + + void set(const URI<>& uri, const std::string_view req_id, const std::span& request) override { + using namespace std::string_view_literals; + using enum detail::FrameType; + auto& con = findConnection(uri); // send message header + detail::PendingRequest req{}; + req.reqId = req_id; + req.requestType = detail::RequestType::SET; + req.data = IoBuffer{request.data(), request.size()}; + req.state = detail::PendingRequest::RequestState::INITIALIZED; + req.uri = uri.str(); + con._pendingRequests.try_emplace(std::string(req_id), std::move(req)); + } + + void subscribe(const URI<>& uri, const std::string_view req_id) override { + using namespace std::string_view_literals; + using enum detail::FrameType; + auto& con = findConnection(uri); + detail::OpenSubscription sub{}; + sub.state = detail::OpenSubscription::SubscriptionState::INITIALIZED; + sub.uri = uri.str(); + std::string req_id_string{req_id}; + char* req_id_end = req_id_string.data() + req_id_string.size(); + sub.reqId = strtol(req_id_string.data(), &req_id_end, 10); + con._subscriptions.try_emplace(std::string(req_id), std::move(sub)); + } + + void unsubscribe(const URI<>& uri, const std::string_view req_id) override { + using namespace std::string_view_literals; + auto& con = findConnection(uri); + con._subscriptions[std::string{req_id}].state = detail::OpenSubscription::SubscriptionState::UNSUBSCRIBING; + CmwLightHeader header; + header.requestType() = std::to_underlying(detail::RequestType::UNSUBSCRIBE); + std::string reqIdString{req_id}; + char* end = reqIdString.data() + req_id.size(); + header.id() = std::strtol(req_id.data(), &end, 10); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send message header"sv, detail::serialiseCmwLight(header)); // send message header + CmwLightRequestContext ctx; + // send requestContext + using enum detail::FrameType; + detail::send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER)); + } + + bool disconnect(detail::Connection& con) const { +#if not defined(__EMSCRIPTEN__) and (not defined(__clang__) or (__clang_major__ >= 16)) + const auto remove = std::ranges::remove_if(_pollItems, [&con](const zmq_pollitem_t& pollItem) { return pollItem.socket == con._socket.zmq_ptr; }); + _pollItems.erase(remove.begin(), remove.end()); +#else + const auto remove = std::remove_if(_pollItems.begin(), _pollItems.end(), [&con](zmq_pollitem_t& pollItem) { return pollItem.socket == con._socket.zmq_ptr; }); + _pollItems.erase(remove, _pollItems.end()); +#endif + con._connectionState = detail::Connection::ConnectionState::DISCONNECTED; + return true; + } + + static bool handleServerReply(mdp::Message& output, detail::Connection& con, const auto currentTime) { + using namespace std::string_view_literals; + if (con._frames.size() < 2 || con._frames.back().data().size() != con._frames.size() - 2 || static_cast(con._frames.back().data()[0]) != detail::FrameType::HEADER) { + throw std::runtime_error(std::format("received malformed response: wrong number of frames({}) or mismatch with frame descriptor({})", con._frames.size(), con._frames.back().size())); + } + // deserialise header frames[1] + IoBuffer data(con._frames[1].data().data(), con._frames[1].data().size()); + DeserialiserInfo info = checkHeaderInfo(data, DeserialiserInfo{}, ProtocolCheck::LENIENT); + CmwLightHeader header; + auto result = opencmw::deserialise(data, header); + + if (con._connectionState == detail::Connection::ConnectionState::CONNECTING2) { + if (header.requestType() == std::to_underlying(detail::RequestType::REPLY)) { + con._connectionState = detail::Connection::ConnectionState::CONNECTED; + con._lastHeartbeatReceived = currentTime; + return true; + } else { + throw std::runtime_error("expected connection reply but got different message"); + } + } + + using enum detail::RequestType; + switch (detail::RequestType{static_cast(header.requestType())}) { + case REPLY: { + auto& request = con._pendingRequests[std::format("{}", header.id())]; + output.arrivalTime = std::chrono::system_clock::now(); /// timePoint < UTC time when the message was sent/received by the client + output.command = mdp::Command::Final; /// Command < command type (GET, SET, SUBSCRIBE, UNSUBSCRIBE, PARTIAL, FINAL, NOTIFY, READY, DISCONNECT, HEARTBEAT) + char* end = &request.reqId.back(); + output.id = std::strtoul(request.reqId.data(), &end, 10); /// std::size_t + output.protocolName = "RDA3"; /// std::string < unique protocol name including version (e.g. 'MDPC03' or 'MDPW03') + std::string clientRequestID = std::format("{}", header.id()); + output.clientRequestID = IoBuffer{clientRequestID.data(), clientRequestID.size()}; /// IoBuffer < stateful: worker mirrors clientRequestID; stateless: worker generates unique increasing IDs (to detect packet loss) + output.topic = URI{request.uri}; /// URI < URI containing at least and optionally parameters + output.serviceName = output.topic.path().value_or("/"); /// std::string < service endpoint name (normally the URI path only), or client source ID (for broker <-> worker messages) + output.data = IoBuffer{con._frames[2].data().data(), con._frames[2].size()}; /// IoBuffer < request/reply body -- opaque binary, e.g. YaS-, CmwLight-, JSON-, or HTML-based + output.error = ""; /// std::string < UTF-8 strings containing error code and/or stack-trace (e.g. "404 Not Found") + return true; + } + case EXCEPTION: { + auto& request = con._pendingRequests[std::format("{}", header.id())]; + output.arrivalTime = std::chrono::system_clock::now(); /// timePoint < UTC time when the message was sent/received by the client + output.command = mdp::Command::Final; /// Command < command type (GET, SET, SUBSCRIBE, UNSUBSCRIBE, PARTIAL, FINAL, NOTIFY, READY, DISCONNECT, HEARTBEAT) + char* end = &request.reqId.back(); + output.id = std::strtoul(request.reqId.data(), &end, 10); /// std::size_t + output.protocolName = "RDA3"; /// std::string < unique protocol name including version (e.g. 'MDPC03' or 'MDPW03') + std::string clientRequestID = std::format("{}", header.id()); + output.clientRequestID = IoBuffer{clientRequestID.data(), clientRequestID.size()}; /// IoBuffer < stateful: worker mirrors clientRequestID; stateless: worker generates unique increasing IDs (to detect packet loss) + output.topic = URI{request.uri}; /// URI < URI containing at least and optionally parameters + output.serviceName = output.topic.path().value_or("/"); /// std::string < service endpoint name (normally the URI path only), or client source ID (for broker <-> worker messages) + output.data = IoBuffer{}; /// IoBuffer < request/reply body -- opaque binary, e.g. YaS-, CmwLight-, JSON-, or HTML-based + IoBuffer errorData(con._frames[2].data().data(), con._frames[2].data().size()); + DeserialiserInfo errorInfo = checkHeaderInfo(errorData, DeserialiserInfo{}, ProtocolCheck::LENIENT); + CmwLightError error; + auto errorResult = opencmw::deserialise(errorData, error); + output.error = std::format("Error in request: {} (AcqStamp:{}, CycleStamp:{}, Type:{})", error.Message, error.ContextAcqStamp, error.ContextCycleStamp, error.Type); /// std::string < UTF-8 strings containing error code and/or stack-trace (e.g. "404 Not Found") + return true; + } + case SUBSCRIBE: { + auto& sub = con._subscriptions[std::format("{}", header.id())]; + sub.replyId = header.options()->sourceId(); + sub.state = detail::OpenSubscription::SubscriptionState::SUBSCRIBED; + sub.backOff = 20ms; // reset back-off + return false; + } + case UNSUBSCRIBE: { + auto subscriptionForUnsub = con._subscriptions.find(std::format("{}", header.id())); + con._subscriptions.erase(subscriptionForUnsub); + return false; + } + case NOTIFICATION_DATA: { + if (auto sub = std::ranges::find_if(con._subscriptions, [&header](auto& pair) { return pair.second.replyId == header.id(); }); sub == con._subscriptions.end()) { + return false; + } else { + auto subscriptionForNotification = sub->second; + output.arrivalTime = std::chrono::system_clock::now(); /// timePoint < UTC time when the message was sent/received by the client + output.command = mdp::Command::Notify; /// Command < command type (GET, SET, SUBSCRIBE, UNSUBSCRIBE, PARTIAL, FINAL, NOTIFY, READY, DISCONNECT, HEARTBEAT) + output.id = static_cast(sub->second.replyId); /// std::size_t + output.protocolName = "RDA3"; /// std::string < unique protocol name including version (e.g. 'MDPC03' or 'MDPW03') + std::string clientRequestID = std::format("{}", subscriptionForNotification.reqId); + output.clientRequestID = IoBuffer{clientRequestID.data(), clientRequestID.size()}; /// IoBuffer < stateful: worker mirrors clientRequestID; stateless: worker generates unique increasing IDs (to detect packet loss) + output.topic = URI{subscriptionForNotification.uri}; /// URI < URI containing at least and optionally parameters + output.serviceName = output.topic.path().value_or("/"); /// std::string < service endpoint name (normally the URI path only), or client source ID (for broker <-> worker messages) + output.data = IoBuffer{con._frames[2].data().data(), con._frames[2].size()}; /// IoBuffer < request/reply body -- opaque binary, e.g. YaS-, CmwLight-, JSON-, or HTML-based + output.error = ""; /// std::string < UTF-8 strings containing error code and/or stack-trace (e.g. "404 Not Found") + return true; + } + } + case NOTIFICATION_EXC: { + std::string replyId; + if (auto subIt = con._subscriptions.find(replyId); subIt == con._subscriptions.end()) { + return false; + } else { + auto subscriptionForNotifyExc = subIt->second; + output.arrivalTime = std::chrono::system_clock::now(); /// timePoint < UTC time when the message was sent/received by the client + output.command = mdp::Command::Notify; /// Command < command type (GET, SET, SUBSCRIBE, UNSUBSCRIBE, PARTIAL, FINAL, NOTIFY, READY, DISCONNECT, HEARTBEAT) + output.id = 0; /// std::size_t + output.protocolName = "RDA3"; /// std::string < unique protocol name including version (e.g. 'MDPC03' or 'MDPW03') + std::string clientRequestID = std::format("{}", subscriptionForNotifyExc.reqId); + output.clientRequestID = IoBuffer{clientRequestID.data(), clientRequestID.size()}; /// IoBuffer < stateful: worker mirrors clientRequestID; stateless: worker generates unique increasing IDs (to detect packet loss) + output.topic = URI{subscriptionForNotifyExc.uri}; /// URI < URI containing at least and optionally parameters + output.serviceName = output.topic.path().value_or("/"); /// std::string < service endpoint name (normally the URI path only), or client source ID (for broker <-> worker messages) + output.data = IoBuffer{}; /// IoBuffer < request/reply body -- opaque binary, e.g. YaS-, CmwLight-, JSON-, or HTML-based + IoBuffer errorData(con._frames[2].data().data(), con._frames[2].data().size()); + DeserialiserInfo errorInfo = checkHeaderInfo(errorData, DeserialiserInfo{}, ProtocolCheck::LENIENT); + CmwLightError error; + auto errorResult = opencmw::deserialise(errorData, error); + output.error = std::format("Error in subscription: {} (AcqStamp:{}, CycleStamp:{}, Type:{})", error.Message, error.ContextAcqStamp, error.ContextCycleStamp, error.Type); /// std::string < UTF-8 strings containing error code and/or stack-trace (e.g. "404 Not Found") + return true; + } + } + case SUBSCRIBE_EXCEPTION: { + auto subForSubExc = con._subscriptions[std::format("{}", header.id())]; + subForSubExc.state = detail::OpenSubscription::SubscriptionState::UNSUBSCRIBED; + subForSubExc.nextTry = currentTime + subForSubExc.backOff; + subForSubExc.backOff *= 2; + // exception during subscription, retrying + output.arrivalTime = std::chrono::system_clock::now(); /// timePoint < UTC time when the message was sent/received by the client + output.command = mdp::Command::Notify; /// Command < command type (GET, SET, SUBSCRIBE, UNSUBSCRIBE, PARTIAL, FINAL, NOTIFY, READY, DISCONNECT, HEARTBEAT) + output.id = static_cast(subForSubExc.reqId); /// std::size_t + output.protocolName = "RDA3"; /// std::string < unique protocol name including version (e.g. 'MDPC03' or 'MDPW03') + std::string clientRequestID = std::format("{}", subForSubExc.reqId); + output.clientRequestID = IoBuffer{clientRequestID.data(), clientRequestID.size()}; /// IoBuffer < stateful: worker mirrors clientRequestID; stateless: worker generates unique increasing IDs (to detect packet loss) + output.clientRequestID = IoBuffer{clientRequestID.data(), clientRequestID.size()}; /// IoBuffer < stateful: worker mirrors clientRequestID; stateless: worker generates unique increasing IDs (to detect packet loss) + output.topic = URI{subForSubExc.uri}; /// URI < URI containing at least and optionally parameters + output.serviceName = output.topic.path().value_or("/"); /// std::string < service endpoint name (normally the URI path only), or client source ID (for broker <-> worker messages) + output.data = IoBuffer{}; /// IoBuffer < request/reply body -- opaque binary, e.g. YaS-, CmwLight-, JSON-, or HTML-based + IoBuffer errorData(con._frames[2].data().data(), con._frames[2].data().size()); + DeserialiserInfo errorInfo = checkHeaderInfo(errorData, DeserialiserInfo{}, ProtocolCheck::LENIENT); + CmwLightError error; + auto errorResult = opencmw::deserialise(errorData, error); + output.error = std::format("Error in subscription: {} (AcqStamp:{}, CycleStamp:{}, Type:{})", error.Message, error.ContextAcqStamp, error.ContextCycleStamp, error.Type); /// std::string < UTF-8 strings containing error code and/or stack-trace (e.g. "404 Not Found") + return true; + } + case SESSION_CONFIRM: { + return false; + } + // These request types should never be returned by the server or are unsupported by this implementation + case GET: + case SET: + case CONNECT: + case EVENT: + default: return false; + } + } + + static bool handleMessage(mdp::Message& output, detail::Connection& con) { + assert(!con._frames.empty() && "this function can only be ever called with at least one frame"); + const auto currentTime = std::chrono::system_clock::now(); + using enum detail::MessageType; + using enum detail::Connection::ConnectionState; + switch (static_cast(con._frames[0].data().at(0))) { + case SERVER_CONNECT_ACK: + if (con._connectionState == CONNECTING1) { + if (con._frames.size() < 2 || con._frames[1].data().empty()) { + throw std::runtime_error("server connect message does not contain required version info"); + } + if (con._frames[1].data() != "1.0.0") { + throw std::runtime_error(std::format("currently only version 1.0.0 is supported, server provided version: {}", con._frames[1].data())); + } + con._connectionState = CONNECTING2; // proceed to step 2 by sending the CLIENT_REQ, REQ_TYPE=CONNECT message + sendConnectRequest(con); + con._lastHeartbeatReceived = currentTime; + con._backoff = 20ms; // reset back-off time + } else { + throw std::runtime_error("received unsolicited SERVER_CONNECT_ACK"); + } + break; + case SERVER_HB: + if (con._connectionState != CONNECTED && con._connectionState != CONNECTING2) { // heart-beat message on an unconnected connection + return false; + } + con._lastHeartbeatReceived = currentTime; + break; + case SERVER_REP: con._lastHeartbeatReceived = currentTime; return handleServerReply(output, con, currentTime); + case CLIENT_CONNECT: + case CLIENT_REQ: + case CLIENT_HB: + default: throw std::runtime_error("Unexpected client message type received from server"); + } + return false; + } + + bool receive(mdp::Message& output) override { + for (auto& con : _connections) { + while (true) { + zmq::MessageFrame frame; + if (const auto byteCountResultId = frame.receive(con._socket, ZMQ_DONTWAIT); !byteCountResultId.isValid() || byteCountResultId.value() < 1) { + break; + } + con._frames.push_back(std::move(frame)); + int64_t more; + size_t moreSize = sizeof(more); + if (!zmq::invoke(zmq_getsockopt, con._socket, ZMQ_RCVMORE, &more, &moreSize)) { + throw std::runtime_error("error checking rcvmore"); + } else if (more == 0) { // the message is complete + const bool received = handleMessage(output, con); + con._frames.clear(); + if (received) { + return true; + } + } + } + } + return false; + } + + // method to be called in regular time intervals to send and verify heartbeats + timePoint housekeeping(const timePoint& now) override { + using ConnectionState = detail::Connection::ConnectionState; + using RequestState = detail::PendingRequest::RequestState; + using namespace std::literals; + using enum detail::OpenSubscription::SubscriptionState; + using enum detail::FrameType; + // handle connection state + for (auto& con : _connections) { + switch (con._connectionState) { + case ConnectionState::DISCONNECTED: + if (con._nextReconnectAttemptTimeStamp <= now) { + if (con._authority.empty()) { + con._connectionState = ConnectionState::NS_LOOKUP; + } else { + connect(con); + } + } + break; + case ConnectionState::NS_LOOKUP: { + auto deviceName = con._deviceName; + if (auto address = _directoryClient.lookup(deviceName); address.has_value()) { + con._authority = address.value(); + connect(con); + } + break; + } + case ConnectionState::CONNECTING1: + case ConnectionState::CONNECTING2: { + if (con._nextReconnectAttemptTimeStamp + _clientTimeout < now) { + // abort this connection attempt and start a new one + } + break; + } + case ConnectionState::CONNECTED: + for (auto& req : con._pendingRequests | std::views::values) { + using enum detail::RequestType; + if (req.state == RequestState::INITIALIZED) { + if (req.requestType == GET && !req.reqId.empty()) { + detail::send(con._socket, ZMQ_SNDMORE, "error sending get frame"sv, std::to_underlying(detail::MessageType::CLIENT_REQ)); + CmwLightHeader msg; + msg.requestType() = std::to_underlying(GET); + char* reqIdEnd = req.reqId.data() + req.reqId.size(); + msg.id() = std::strtol(req.reqId.data(), &reqIdEnd, 10); + msg.sessionId() = detail::createClientId(); + URI uri{req.uri}; + msg.device() = uri.path()->substr(1, uri.path()->find('/', 1) - 1); + msg.property() = uri.path()->substr(uri.path()->find('/', 1) + 1); + msg.options() = std::make_unique(); + msg.updateType() = std::to_underlying(detail::UpdateType::NORMAL); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send message header"sv, detail::serialiseCmwLight(msg)); // send message header + if (const auto& params = uri.queryParamMap(); params.contains("ctx")) { + CmwLightRequestContext ctx; + for (const auto& [key, value] : params) { + if (key == "ctx") { + ctx.selector() = value.value_or("FAIR.SELECTOR.ALL"); + } else { + ctx.filters().try_emplace(key, value.value_or("")); + } + } + IoBuffer buffer{}; + serialise(buffer, ctx); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send context frame"sv, std::move(buffer)); // send requestContext + detail::send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER, BODY_REQUEST_CONTEXT)); + } else { + detail::send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER)); + } + req.state = RequestState::WAITING; + } else if (req.requestType == SET) { + detail::send(con._socket, ZMQ_SNDMORE, "error sending get frame"sv, std::to_underlying(detail::MessageType::CLIENT_REQ)); + CmwLightHeader msg; + msg.requestType() = std::to_underlying(SET); + char* reqIdEnd = req.reqId.data() + req.reqId.size(); + msg.id() = std::strtol(req.reqId.data(), &reqIdEnd, 10); + msg.sessionId() = detail::createClientId(); + URI uri{req.uri}; + msg.device() = uri.path()->substr(1, uri.path()->find('/', 1) - 1); + msg.property() = uri.path()->substr(uri.path()->find('/', 1) + 1); + msg.updateType() = std::to_underlying(detail::UpdateType::NORMAL); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send message header"sv, detail::serialiseCmwLight(msg)); // send message header + if (auto params = uri.queryParamMap(); params.contains("ctx")) { + CmwLightRequestContext ctx; + ctx.selector() = params["ctx"].value_or("FAIR.SELECTOR.ALL"); + for (auto& [key, value] : params) { + if (key == "ctx") { + continue; + } + ctx.filters().try_emplace(key, value.value_or("")); + } + IoBuffer buffer{}; + serialise(buffer, ctx); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send context frame"sv, std::move(buffer)); // send requestContext + detail::send(con._socket, ZMQ_SNDMORE, "failed to send data frame"sv, std::move(req.data)); // send requestContext + detail::send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER, BODY_REQUEST_CONTEXT, BODY)); + } else { + detail::send(con._socket, ZMQ_SNDMORE, "failed to send data frame"sv, std::move(req.data)); // send requestContext + detail::send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER, BODY)); + } + req.state = RequestState::WAITING; + } + } + } + for (auto& sub : con._subscriptions | std::views::values) { + if (sub.state == INITIALIZED) { + detail::send(con._socket, ZMQ_SNDMORE, "error sending client req frame"sv, std::to_underlying(detail::MessageType::CLIENT_REQ)); + URI uri{sub.uri}; + CmwLightHeader header; + header.id() = sub.reqId; + header.device() = uri.path()->substr(1, uri.path()->find('/', 1) - 1); + header.property() = uri.path()->substr(uri.path()->find('/', 1) + 1); + header.requestType() = std::to_underlying(detail::RequestType::SUBSCRIBE); + header.options() = std::make_unique(); + header.sessionId() = detail::createClientId(); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send message header"sv, detail::serialiseCmwLight(header)); // send message header + auto queryParams = uri.queryParamMap(); + CmwLightRequestContext ctx; + for (auto& [key, value] : queryParams) { + if (key == "ctx") { + ctx.selector() = queryParams.contains("ctx") ? queryParams["ctx"].value_or("") : ""; + } else { + ctx.filters().try_emplace(key, value.value_or("")); + } + } + IoBuffer buffer{}; + serialise(buffer, ctx); + detail::send(con._socket, ZMQ_SNDMORE, "failed to send context frame"sv, std::move(buffer)); // send requestContext + detail::send(con._socket, 0, "failed to send descriptor frame"sv, descriptorToString(HEADER, BODY_REQUEST_CONTEXT)); + con._lastHeartBeatSent = now; + sub.state = SUBSCRIBING; + } else if (sub.state == UNSUBSCRIBING) { + // TODO resend unsubscribe request on timeout, forcefully remove subscription after retries + } + } + if (con._lastHeartBeatSent < now - HEARTBEAT_INTERVAL) { + detail::send(con._socket, 0, "error sending heart-beat"sv, std::to_underlying(detail::MessageType::CLIENT_HB)); + con._lastHeartBeatSent = now; + } + if (con._lastHeartbeatReceived < now - HEARTBEAT_INTERVAL * 3) { + disconnect(con); + con._nextReconnectAttemptTimeStamp = now + con._backoff; + con._backoff = std::min(2 * con._backoff, std::chrono::duration_cast(MAX_BACKOFF)); + std::println("Missed 3 heartbeats -> connection seems to be broken"); + } + break; // do nothing + } + } + return now + _clientTimeout / 2; + } +}; + +/* + * Implementation of the Majordomo client protocol. Spawns a single thread which controls all client connections and sockets. + * A dispatcher thread reads the requests from the command ring buffer and dispatches them to the zeromq poll loop using an inproc socket pair. + */ +class CmwLightClientCtx : public ClientBase { + using timeUnit = std::chrono::milliseconds; + std::unordered_map, std::unique_ptr> _clients; + const zmq::Context& _zctx; + DirectoryLightClient _nameserver; + zmq::Socket _control_socket_send; + zmq::Socket _control_socket_recv; + std::jthread _poller; + std::vector _pollitems{}; + std::unordered_map _requests; + std::unordered_map _subscriptions; + timeUnit _timeout; + std::string _clientId; + std::size_t _request_id = 1; + +public: + explicit CmwLightClientCtx(const zmq::Context& zeromq_context, std::string nameserver, const timeUnit timeout = 100ms, std::string clientId = "") : _zctx{zeromq_context}, _nameserver{DirectoryLightClient{std::move(nameserver)}}, _control_socket_send(zeromq_context, ZMQ_PAIR), _control_socket_recv(zeromq_context, ZMQ_PAIR), _timeout(timeout), _clientId(std::move(clientId)) { + zmq::invoke(zmq_bind, _control_socket_send, "inproc://cmwlightclientControlSocket").onFailure("could not bind control socket"); + _pollitems.push_back({.socket = _control_socket_recv.zmq_ptr, .fd = 0, .events = ZMQ_POLLIN, .revents = 0}); + _poller = std::jthread([this](const std::stop_token& stoken) { this->poll(stoken); }); + } + + std::vector protocols() override { + return {"rda3", "rda3tcp"}; // rda3 protocol, if transport is unspecified, tcp is used if authority contains a port + } + + std::unique_ptr& getClient(const URI<>& uri) { + auto baseUri = URI<>::factory(uri).setQuery({}).path("").fragment("").build(); + if (_clients.contains(baseUri)) { + return _clients.at(baseUri); + } + auto [it, ins] = _clients.emplace(baseUri, createClient(_nameserver)); + if (!ins) { + throw std::logic_error("could not insert client into client list"); + } + return it->second; + } + + std::unique_ptr createClient(DirectoryLightClient& nameserver) { return std::make_unique(_zctx, _pollitems, nameserver, _timeout, _clientId); } + + void stop() override { + _poller.request_stop(); + _poller.join(); + } + + void request(Command cmd) override { + std::size_t req_id = 0; + if (cmd.command == mdp::Command::Get || cmd.command == mdp::Command::Set) { + req_id = _request_id++; + _requests.try_emplace(req_id, Request{.uri = cmd.topic, .callback = std::move(cmd.callback), .timestamp_received = cmd.arrivalTime}); + } else if (cmd.command == mdp::Command::Subscribe) { + req_id = _request_id++; + std::string clientRequestId = std::format("{}", req_id); + _subscriptions.try_emplace(clientRequestId, Subscription{.uri = cmd.topic, .callback = std::move(cmd.callback), .timestamp_received = cmd.arrivalTime, .reqId = req_id}); + } else if (cmd.command == mdp::Command::Unsubscribe) { + std::string clientRequestId{}; + for (auto& [reqId, sub] : _subscriptions) { + if (sub.uri == cmd.topic) { + clientRequestId = reqId; + break; + } + } + if (const auto subIt = _subscriptions.find(clientRequestId); subIt != _subscriptions.end()) { + req_id = subIt->second.reqId; + _subscriptions.erase(subIt); + }; + } + sendCmd(cmd.topic, cmd.command, req_id, cmd.data); + } + + DirectoryLightClient& nameserverClient() { return _nameserver; }; + +private: + void sendCmd(const URI<>& uri, mdp::Command commandType, const std::size_t req_id, IoBuffer data = {}) const { + const bool isSet = commandType == mdp::Command::Set; + zmq::MessageFrame cmdType{std::string{static_cast(commandType)}}; + cmdType.send(_control_socket_send, ZMQ_SNDMORE).assertSuccess(); + zmq::MessageFrame reqId{std::to_string(req_id)}; + reqId.send(_control_socket_send, ZMQ_SNDMORE).assertSuccess(); + zmq::MessageFrame endpoint{std::string(uri.str())}; + endpoint.send(_control_socket_send, isSet ? ZMQ_SNDMORE : 0).assertSuccess(); + if (isSet) { + zmq::MessageFrame dataframe{std::move(data)}; + dataframe.send(_control_socket_send, 0).assertSuccess(); + } + } + + void handleRequests() { + zmq::MessageFrame cmd; + zmq::MessageFrame reqId; + zmq::MessageFrame endpoint; + while (cmd.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) { + if (!reqId.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) { + throw std::logic_error("invalid request received: failure receiving message"); + } + if (!endpoint.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) { + throw std::logic_error("invalid request received: invalid message contents"); + } + URI uri{std::string(endpoint.data())}; + const auto& client = getClient(uri); + if (cmd.data().size() != 1) { + throw std::logic_error("invalid request received: wrong number of frames"); + } else if (cmd.data()[0] == std::to_underlying(mdp::Command::Get)) { + client->get(uri, reqId.data()); + } else if (cmd.data()[0] == std::to_underlying(mdp::Command::Set)) { + zmq::MessageFrame data; + if (!data.receive(_control_socket_recv, ZMQ_DONTWAIT).isValid()) { + throw std::logic_error("missing set data"); + } + client->set(uri, reqId.data(), std::span(data.data().data(), data.data().size())); + } else if (cmd.data()[0] == std::to_underlying(mdp::Command::Subscribe)) { + client->subscribe(uri, reqId.data()); + } else if (cmd.data()[0] == std::to_underlying(mdp::Command::Unsubscribe)) { + client->unsubscribe(uri, reqId.data()); + } else { + throw std::logic_error("invalid request received"); // messages always consist of 2 frames + } + } + } + + void poll(const std::stop_token& stoken) { + using enum mdp::Command; + thread::setThreadName("CmwLightPollerThread"); + auto nextHousekeeping = std::chrono::system_clock::now(); + zmq::invoke(zmq_connect, _control_socket_recv, "inproc://cmwlightclientControlSocket").assertSuccess(); + while (!stoken.stop_requested() && zmq::invoke(zmq_poll, _pollitems.data(), static_cast(_pollitems.size()), 200)) { + if (auto now = std::chrono::system_clock::now(); nextHousekeeping < now) { + nextHousekeeping = housekeeping(now); + // expire old subscriptions/requests/connections + } + handleRequests(); + for (const auto& client : _clients | std::views::values) { + mdp::Message receivedEvent; + while (client->receive(receivedEvent)) { + if (auto now = std::chrono::system_clock::now(); nextHousekeeping < now) { // perform housekeeping duties periodically + nextHousekeeping = housekeeping(now); + } + if (receivedEvent.command == Invalid) { // Protocol internal messages, which require further receives but no publishing + continue; + } + if (std::string clientRequestId{reinterpret_cast(receivedEvent.clientRequestID.data()), receivedEvent.clientRequestID.size()}; receivedEvent.command == Notify && _subscriptions.contains(clientRequestId)) { + _subscriptions.at(clientRequestId).callback(receivedEvent); // callback + } + if (receivedEvent.command == Final && _requests.contains(receivedEvent.id)) { + _requests.at(receivedEvent.id).callback(receivedEvent); // callback + _requests.erase(receivedEvent.id); + } + } + } + } + } + + timePoint housekeeping(const timePoint now) const { + timePoint next = now + _timeout; + for (const auto& client : _clients | std::views::values) { + next = std::min(next, client->housekeeping(now)); + } + return next; + } +}; +} // namespace opencmw::client::cmwlight +#endif // OPENCMW_CPP_CMWLIGHTCLIENT_HPP diff --git a/blocklib/opencmw/include/fair/opencmw/cmwlight/DirectoryLightClient.hpp b/blocklib/opencmw/include/fair/opencmw/cmwlight/DirectoryLightClient.hpp new file mode 100644 index 00000000..a8c16ec9 --- /dev/null +++ b/blocklib/opencmw/include/fair/opencmw/cmwlight/DirectoryLightClient.hpp @@ -0,0 +1,239 @@ +#ifndef OPENCMW_CPP_DIRECTORYLIGHTCLIENT_HPP +#define OPENCMW_CPP_DIRECTORYLIGHTCLIENT_HPP + +#include +#include +#include +#include + +#include + +#include + +#include + +namespace opencmw::client::cmwlight { +struct NameserverReplyLocation { + std::string domain; + std::string endpoint; +}; +struct NameserverReplyServer { + std::string name; + NameserverReplyLocation location; +}; + +struct NameserverReplyResource { + std::string name; + NameserverReplyServer server; +}; + +struct NameserverReply { + std::vector resources; +}; +} // namespace opencmw::client::cmwlight +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::NameserverReplyLocation, domain, endpoint) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::NameserverReplyServer, name, location) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::NameserverReplyResource, name, server) +ENABLE_REFLECTION_FOR(opencmw::client::cmwlight::NameserverReply, resources) + +/* + * Implements the CMW Directory Server lookup + * + * minimal API usage example: + * curl ${CMW_NAMESERVER}/api/v1/devices/search --json '{ "proxyPreferred" : true, "domains" : [ ], "directServers" : [ ], "redirects" : { }, "names" : [ "GSITemplateDevice" ]}' + * {"resources":[{"name":"GSITemplateDevice","server":{"name":"GSITemplate_DU.vmla017","location":{"domain":"RDA3","endpoint":"9#Address:#string#19#tcp:%2F%2Fvmla017:36725#ApplicationId:#string#124#app=GSITemplate_DU;uid=root;host=vmla017;pid=398;os=Linux%2D6%2E6%2E111%2Drt31%2Dyocto%2Dpreempt%2Drt;osArch=64bit;appArch=64bit;lang=C%2B%2B;#Language:#string#3#C%2B%2B#Name:#string#22#GSITemplate_DU%2Evmla017#Pid:#int#398#ProcessName:#string#14#GSITemplate_DU#StartTime:#long#1771235547903#UserName:#string#4#root#Version:#string#5#5%2E1%2E1"}}}]} + */ + +namespace opencmw::client::cmwlight { +class DirectoryLightClient { + using EndpointMap = std::map, std::expected, std::expected>, std::less<>>; + std::map> cache; + std::mutex mutex; + std::deque pendingLookups; + std::string nameserver; + +public: + explicit DirectoryLightClient(std::string _nameserver) : nameserver(std::move(_nameserver)) {} + + ~DirectoryLightClient() { cpr::GlobalThreadPool::GetInstance()->Stop(); }; + + std::optional lookup(const std::string_view name, const bool useCache = true) { + { + std::lock_guard lock{mutex}; + if (useCache) { + if (const auto& res = cache.find(name); res != cache.end()) { + if (!res->second.empty()) { + return res->second; + } else { + return {}; // request was already sent + } + } + } + cache[std::string{name}] = ""; + } + triggerRequest(name); + return {}; + } + + void addStaticLookup(const std::string& deviceName, const std::string_view address) { + std::lock_guard lock{mutex}; + cache[deviceName] = address; + }; + + static std::expected urlDecode(std::string_view str) { + std::string ret; + ret.reserve(str.length()); + const std::size_t len = str.length(); + for (std::size_t i = 0; i < len; i++) { + if (str[i] != '%') { + if (str[i] == '+') { + ret += ' '; + } else { + ret += str[i]; + } + } else if (i + 2 < len) { + auto isHex = [](char c) { return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F'); }; + auto toHex = [](char c) { + if (c >= '0' && c <= '9') { + return c - '0'; + } + if (c >= 'a' && c <= 'f') { + return c - 'a' + 10; + } + if (c >= 'A' && c <= 'F') { + return c - 'A' + 10; + } + return 0; + }; + if (!isHex(str.at(i + 1)) || !isHex(str.at(i + 2))) { + return std::unexpected(std::format("invalid hexadecimal encoding at position {} in '{}'", i, str)); + } + const auto ch = static_cast('\x10' * toHex(str.at(i + 1)) + toHex(str.at(i + 2))); + ret += ch; + i = i + 2; + } else { + return std::unexpected(std::format("incomplete hexadecimal encoding at position {} in '{}'", i, str)); + } + } + return ret; + } + + static std::expected parseEndpointString(std::string_view sizeStr, std::string_view valueStr) { + std::string str2{sizeStr}; + char* parsed; + std::size_t size = std::strtoull(str2.data(), &parsed, 10); + if (parsed != std::to_address(str2.end())) { + return std::unexpected(std::format("Error parsing string size: {}", sizeStr)); + } + if (const auto result = urlDecode(valueStr); result.has_value()) { + if (result.value().size() == size) { + return result; + } else { + return std::unexpected(std::format("invalid size {} for string value {}", size, valueStr)); + } + } else { + return result; + } + } + + static std::expected parseEndpointInt(std::string_view str) { return std::atoi(str.data()); } + + static std::expected parseEndpointLong(std::string_view str) { + std::string str2{str}; + auto parsed = std::to_address(str2.end()); + long number = std::strtol(str2.data(), &parsed, 10); + if (parsed != std::to_address(str2.end())) { + return std::unexpected(std::format("Error parsing long: {}", str)); + } + return number; + } + + static EndpointMap parseEndpoint(std::string_view endpoint) { + using namespace std::literals; + auto tokens = std::views::split(endpoint, "#"sv); + if (tokens.empty()) { + return {}; + } + auto iterator = tokens.begin(); + std::string fieldCountString = {(*iterator).data(), (*iterator).size()}; + char* end = to_address(fieldCountString.end()); + std::size_t fieldCount = std::strtoull(fieldCountString.data(), &end, 10); + ++iterator; + EndpointMap result; + std::size_t n = 0; + while (n < fieldCount && iterator != tokens.end()) { + std::string_view fieldNameView{&(*iterator).front(), (*iterator).size()}; + std::string fieldname{fieldNameView.substr(0, fieldNameView.size() - 1)}; + if (++iterator == tokens.end()) { + result.try_emplace("parserError", std::expected(std::unexpected(std::format("incomplete field {} in endpoint {}", fieldname, endpoint)))); + break; + } + std::string type{std::string_view{&(*iterator).front(), (*iterator).size()}}; + if (++iterator == tokens.end()) { + result.try_emplace("parserError", std::expected(std::unexpected(std::format("incomplete field {} in endpoint {}", fieldname, endpoint)))); + break; + } + std::string_view valueView{&(*iterator).front(), (*iterator).size()}; + if (type == "string") { + if (++iterator == tokens.end()) { + result.try_emplace("parserError", std::expected(std::unexpected(std::format("incomplete field {} in endpoint {}", fieldname, endpoint)))); + break; + } + auto value = parseEndpointString(valueView, std::string_view{&(*iterator).front(), (*iterator).size()}); + result.try_emplace(fieldname, value); + } else if (type == "int") { + auto value = parseEndpointInt(valueView); + result.try_emplace(fieldname, value); + } else if (type == "long") { + auto value = parseEndpointLong(valueView); + result.try_emplace(fieldname, value); + } else { + result.try_emplace(fieldname, std::expected(std::unexpected(std::format("invalid field type {} in '{}'", type, endpoint)))); + } + ++iterator; + ++n; + } + if (iterator != tokens.end()) { + result.try_emplace("parserError", std::expected(std::unexpected(std::format("expected {} fields in '{}', but there are more fields", fieldCount, endpoint)))); + } + return result; + } + + static std::map> parseNameserverReply(const std::string& reply) { + IoBuffer buffer{reply.data(), reply.size()}; + NameserverReply replyObj; + auto res = opencmw::deserialise(buffer, replyObj); + for (const auto& [name, server] : replyObj.resources) { + if (server.location.domain != "RDA3") { + continue; + } + return {{name, parseEndpoint(server.location.endpoint)}}; + } + return {}; + } + + void triggerRequest(std::string_view name) { + using namespace std::literals; + const std::string requestData = std::format(R"""({{ "proxyPreferred" : true, "domains" : [ ], "directServers" : [ ], "redirects" : {{ }}, "names" : [ "{}" ]}})""", name); + const auto uri = URI<>::factory(URI(std::string(nameserver))).path("/api/v1/devices/search").build(); + std::string deviceName{name}; + cpr::PostCallback( + [this, deviceName](const cpr::Response& response) { + if (response.status_code == 200) { + for (auto& [nameKey, data] : parseNameserverReply(response.text)) { + if (nameKey != deviceName) { + continue; + } + std::lock_guard lock{mutex}; + cache[std::string{nameKey}] = std::get>(data.at("Address")).value(); + return; + } + } + std::lock_guard lock{mutex}; + cache.erase(deviceName); // lookup failed, do not cache failure + }, + cpr::Url{uri.str()}, cpr::Body{requestData}, cpr::Header{{"Content-Type", "application/json"}}); + } +}; +} // namespace opencmw::client::cmwlight +#endif // OPENCMW_CPP_DIRECTORYLIGHTCLIENT_HPP diff --git a/blocklib/opencmw/test/CMakeLists.txt b/blocklib/opencmw/test/CMakeLists.txt new file mode 100644 index 00000000..1856bbdf --- /dev/null +++ b/blocklib/opencmw/test/CMakeLists.txt @@ -0,0 +1,33 @@ +function(add_ut_test_tool TEST_NAME) # for manual usage, not run by ctest + add_executable(${TEST_NAME} ${TEST_NAME}.cpp) + target_include_directories( + ${TEST_NAME} + PRIVATE ${CMAKE_BINARY_DIR}/include ${CMAKE_CURRENT_BINARY_DIR} + ${CMAKE_CURRENT_SOURCE_DIR}) + target_link_libraries( + ${TEST_NAME} + PRIVATE gnuradio-core + gr-testing + gr-basic + fair-opencmw + ut + gnuradio-blocklib-core + GrBasicBlocksShared) +endfunction() + +function(add_ut_test TEST_NAME) + add_ut_test_tool(${TEST_NAME}) + add_test(NAME ${TEST_NAME} COMMAND ${CMAKE_CROSSCOMPILING_EMULATOR} + ${CMAKE_CURRENT_BINARY_DIR}/${TEST_NAME}) +endfunction() + +add_ut_test(qa_OpenCmwClient) +add_ut_test(qa_OpenCmwSource) + +add_custom_target( + copy-rda3-jar ALL + COMMAND + ${CMAKE_COMMAND} -E copy ${CMAKE_CURRENT_SOURCE_DIR}/DemoRda3Server.jar + ${CMAKE_CURRENT_BINARY_DIR}) +add_dependencies(qa_OpenCmwClient copy-rda3-jar) +add_dependencies(qa_OpenCmwSource copy-rda3-jar) diff --git a/blocklib/opencmw/test/DemoRda3Server.jar b/blocklib/opencmw/test/DemoRda3Server.jar new file mode 100644 index 00000000..4197153c Binary files /dev/null and b/blocklib/opencmw/test/DemoRda3Server.jar differ diff --git a/blocklib/opencmw/test/ProcessHelper.hpp b/blocklib/opencmw/test/ProcessHelper.hpp new file mode 100644 index 00000000..da392e1c --- /dev/null +++ b/blocklib/opencmw/test/ProcessHelper.hpp @@ -0,0 +1,47 @@ +#ifndef GR_DIGITIZERS_PROCESSHELPER_HPP +#define GR_DIGITIZERS_PROCESSHELPER_HPP +#include +#include +#include +#include +#include +#include +#include +#include + +class Process { +public: + explicit Process(const std::vector& args) { + if (const pid_t pid = fork(); pid == -1) { + throw std::runtime_error("fork failed"); + } else if (pid == 0) { // Child process: build argv and exec + std::vector argv; + for (const auto& arg : args) { + argv.push_back(const_cast(arg.c_str())); + } + argv.push_back(nullptr); + execvp(argv[0], argv.data()); // replace the forked process with the requested one + std::exit(EXIT_FAILURE); + } else { // Parent process: store pid + pid_ = pid; + } + } + + ~Process() { + if (pid_ > 0) { + kill(pid_, SIGTERM); + waitpid(pid_, nullptr, 0); + } + } + + // Prevent copying/moving + Process(const Process&) = delete; + Process& operator=(const Process&) = delete; + Process(Process&&) = delete; + Process& operator=(Process&&) = delete; + +private: + pid_t pid_ = -1; +}; + +#endif // GR_DIGITIZERS_PROCESSHELPER_HPP diff --git a/blocklib/opencmw/test/qa_OpenCmwClient.cpp b/blocklib/opencmw/test/qa_OpenCmwClient.cpp new file mode 100644 index 00000000..24523cff --- /dev/null +++ b/blocklib/opencmw/test/qa_OpenCmwClient.cpp @@ -0,0 +1,484 @@ +#include +#include + +#include + +#include + +#include +#include + +#include "ProcessHelper.hpp" + +/** + * Most tests in this suite require a running Test FESA Server which is registered to a Nameserver. + * Additionally, the following Environment Variables have to be set: + * - CMW_NAMESERVER=http://host:port - URL to the Nameserver to query + * - CMW_DEVICE_ADDRESS=tcp://host:port - URL where an instance of the GSITemplateDevice FESA class is running + * This can be obtained via nameserver lookup, but has to be set if the device needs ssh tunnelling with ssh -L local_port:device:port + * Use the output of the nameserver lookup test to determine the hostname and port of the device to set up the ssh tunnel. + * + * FESA Test Device: + * - vmla017 + * - DEV nameserver + * - Device: GSITemplateDevice + * - servername: GSITemplate_DU.vmla017 + * - Properties: + * - Acquisition: + * - Version + * - ModuleStatus + * - Status + * - Acquisition (mux) + * - Voltage: generates random values for beam processes 1 and 2 every 2 seconds + * - Setting: + * - Power + * - Init + * - Reset + * - Setting (mux) + * - Voltage: determines the acquisition values + */ + +struct TestSettings { + double voltage = 42.23; +}; +ENABLE_REFLECTION_FOR(TestSettings, voltage); + +static bool waitFor(const std::atomic& counter, const int expectedValue, const auto timeout) { + const auto start = std::chrono::system_clock::now(); + while (counter.load() < expectedValue && std::chrono::system_clock::now() - start < timeout) { + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } + return counter.load() == expectedValue; +} + +static bool waitFor(const std::atomic& counter, const int expectedValue) { + using namespace std::literals; + return waitFor(counter, expectedValue, 1s); +} + +const boost::ut::suite OpenCmwTests = [] { + using namespace boost::ut; + using namespace std::literals; + + "ParseNameserverReply"_test = [] { + std::println("# test parsing nameserver reply - starting"); + const std::string nameserverReply = R"""({"resources":[{"name":"GGGG001","server":{"name":"TestDU2.srv001","location":{"domain":"RDA3","endpoint":"9#Address:#string#17#tcp:%2F%2Fsrv001:1234#ApplicationId:#string#108#app=TestDU2;uid=root;host=srv001;pid=1977;os=Linux%2D3%2E10%2E101%2Drt111%2Dpc002;osArch=64bit;appArch=64bit;lang=C%2B%2B;#Language:#string#3#C%2B%2B#Name:#string#14#TestDU2%2Esrv001#Pid:#int#1977#ProcessName:#string#7#TestDU2#StartTime:#long#1769161872191#UserName:#string#4#root#Version:#string#5#3%2E1%2E0"}}}]})"""; + const auto result = opencmw::client::cmwlight::DirectoryLightClient::parseNameserverReply(nameserverReply); + expect(std::get>(result.at("GGGG001").at("Address")).value() == "tcp://srv001:1234"); + std::println("# test parsing nameserver reply - finished"); + }; + + "Query rda3 directory server/nameserver"_test = [] { + std::println("# query rda3 nameserver - starting"); + const auto env_nameserver = std::getenv("CMW_NAMESERVER"); + if (env_nameserver == nullptr) { + std::println("skipping BasicCmwLight example test as it relies on the availability of network infrastructure."); + return; // skip test + } + const std::string nameserver{env_nameserver}; + const opencmw::zmq::Context zctx{}; + opencmw::client::cmwlight::DirectoryLightClient nameserverClient{nameserver}; + std::optional result = nameserverClient.lookup("GSITemplateDevice"); + expect(!result.has_value()); // check that the device is originally not in the nameserver's cache + std::chrono::system_clock::time_point start = std::chrono::system_clock::now(); + while (!(result = nameserverClient.lookup("GSITemplateDevice")).has_value() && std::chrono::system_clock::now() - start < 1s) { + std::this_thread::sleep_for(1ms); + } + expect(result.has_value() && result.value().starts_with("tcp://vmla")); + // check that later lookups immediately return the cached result + result = nameserverClient.lookup("GSITemplateDevice"); + expect(result.has_value() && result.value().starts_with("tcp://vmla")); + std::println("Lookup result: GSITemplateDevice -> {} ", result.value_or("nameserver lookup failed")); + std::println("# query rda3 nameserver - finished"); + }; + + "CmwLightClientGetLocal"_test = [] { + std::println("# Test rda3 get/set - starting"); + Process rda3Server({"java", "-jar", "./DemoRda3Server.jar", "-cfg", "--cmw.rda3.transport.server.port=12345", "-sleep", "1500"}); + using namespace opencmw; + using namespace std::literals; + const std::string DEVICE_NAME = "GSITemplateDevice"; + const std::string STATUS_PROPERTY = "/GSITemplateDevice/Status"; + const std::string SETTING_PROPERTY = "/GSITemplateDevice/Setting"; + const std::string ACQUISITION_PROPERTY = "/GSITemplateDevice/Acquisition"; + const std::string SELECTOR = "FAIR.SELECTOR.C=3:S=1:P=1:T=300"; + + std::string nameserver = "http://localhost:7500"; // this is actually never queried since all requests in this test provide the url authority + const zmq::Context zctx{}; + std::vector> clients; + auto cmwlightClient = std::make_unique(zctx, nameserver, 100ms, "testclient"); + std::string deviceHost = "rda3://localhost:12345"; + clients.emplace_back(std::move(cmwlightClient)); + client::ClientContext clientContext{std::move(clients)}; + // send some requests + { // non-multiplexed get request + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(STATUS_PROPERTY).build(); + std::atomic getReceived{0}; + clientContext.get(endpoint, [&getReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "get should have succeeded"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 13_i) << "wrong number of field in subscription update"; + ++getReceived; + } + }); + expect(waitFor(getReceived, 1)) << "timed-out waiting for get response"; + } + { // multiplexed get request + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(ACQUISITION_PROPERTY).addQueryParameter("ctx", SELECTOR).build(); + std::atomic getReceived{0}; + clientContext.get(endpoint, [&getReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "get should have succeeded"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 12_i) << "wrong number of field in subscription update"; + ++getReceived; + } + }); + expect(waitFor(getReceived, 1)) << "timed-out waiting for get response"; + } + { // multiplexed set request + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(SETTING_PROPERTY).addQueryParameter("ctx", SELECTOR).build(); + std::atomic setReceived{0}; + TestSettings testSettings{}; + IoBuffer data{}; + opencmw::serialise(data, testSettings); + clientContext.set( + endpoint, + [&setReceived](const mdp::Message& message) { + if (!message.error.empty()) { + std::println("Error: {}", message.error); + expect(false) << "set should have succeeded"; + } else { + ++setReceived; + } + }, + std::move(data)); + expect(waitFor(setReceived, 1)) << "timed-out waiting for set response"; + } + { // error-case: multiplexed get request with missing selector + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(SETTING_PROPERTY).build(); + std::atomic getError{0}; + clientContext.get(endpoint, [&getError](const mdp::Message& message) { + if (!message.error.empty()) { + expect(message.error.contains("Access point 'GSITemplateDevice/Setting' needs a selector")); + ++getError; + } else { + expect(false) << "get should have failed"; + } + }); + expect(waitFor(getError, 1)) << "timed-out waiting for get error"; + } + { // error-case: non-existent device property + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path("GSITemplateDevice/NON_EXISTENT_PROPERTY").build(); + std::atomic getError{0}; + clientContext.get(endpoint, [&getError](const mdp::Message& message) { + if (!message.error.empty()) { + std::println("Error: {}", message.error); + expect(message.error.contains("Error in request: UNKNOWN_PROPERTY: Device 'GSITemplateDevice' has no property named 'NON_EXISTENT_PROPERTY'")); + ++getError; + } else { + expect(false) << "get should have failed"; + } + }); + expect(waitFor(getError, 1)) << "timed-out waiting for get error"; + } + std::println("# Test rda3 get/set - finished"); + }; + + "CmwLightClientSubscribeLocal"_test = [] { + std::println("# test rda3 subscribe - starting"); + Process rda3Server({"java", "-jar", "./DemoRda3Server.jar", "-cfg", "--cmw.rda3.transport.server.port=12345", "-sleep", "1500"}); + using namespace opencmw; + using namespace std::literals; + const std::string DEVICE_NAME = "GSITemplateDevice"; + const std::string STATUS_PROPERTY = "/GSITemplateDevice/Status"; + const std::string ACQUISITION_PROPERTY = "/GSITemplateDevice/Acquisition"; + const std::string SELECTOR = "FAIR.SELECTOR.P=1"; + + std::string nameserver = "http://localhost:7500"; + const zmq::Context zctx{}; + std::vector> clients; + auto cmwlightClient = std::make_unique(zctx, nameserver, 100ms, "testclient"); + std::string deviceHost = "rda3://localhost:12345"; + clients.emplace_back(std::move(cmwlightClient)); + client::ClientContext clientContext{std::move(clients)}; + { // Subscribe non-multiplexed + std::atomic subscriptionUpdatesReceived{0}; + auto subscriptionEndpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(STATUS_PROPERTY).build(); + clientContext.subscribe(subscriptionEndpoint, [&subscriptionUpdatesReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "subscription should not notify exceptions"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 13_i) << "wrong number of field in subscription update"; + ++subscriptionUpdatesReceived; + } + }); + expect(waitFor(subscriptionUpdatesReceived, 2, 5s)) << "timed-out waiting for subscription updates"; // property gets notified every 2 seconds + // Check that subscription updates stop after unsubscribing + clientContext.unsubscribe(subscriptionEndpoint); + std::this_thread::sleep_for(200ms); // get a few subscription updates + int subscriptionUpdatesAfterUnsubscribe = subscriptionUpdatesReceived; + std::this_thread::sleep_for(5s); // get a few subscription updates + expect(eq(subscriptionUpdatesReceived.load(), subscriptionUpdatesAfterUnsubscribe)); + } + { // error case: subscribe to multiplexed without selector + std::atomic subscriptionUpdateError{0}; + auto subscriptionEndpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(ACQUISITION_PROPERTY).build(); + clientContext.subscribe(subscriptionEndpoint, [&subscriptionUpdateError](const mdp::Message& message) { + if (!message.error.empty()) { + expect(message.error.contains("Access point 'GSITemplateDevice/Acquisition' needs a selector")); + ++subscriptionUpdateError; + } else { + expect(false) << "invalid subscription should not notify updates"; + } + }); + expect(waitFor(subscriptionUpdateError, 1, 1s)) << "timed-out waiting for subscription error"; + } + { // subscribe to multiplexed property + std::atomic subscriptionUpdatesReceived{0}; + auto subscriptionEndpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(ACQUISITION_PROPERTY).addQueryParameter("ctx", SELECTOR).build(); + std::println("Subscribing to acquisition property with url {}", subscriptionEndpoint); + clientContext.subscribe(subscriptionEndpoint, [&subscriptionUpdatesReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "subscription should not notify exceptions"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 12_i) << "wrong number of fields in subscription update"; + ++subscriptionUpdatesReceived; + } + }); + expect(waitFor(subscriptionUpdatesReceived, 2, 5s)) << "timed-out waiting for subscription updates"; // property gets notified every 2 seconds + // Check that subscription updates stop after unsubscribing + clientContext.unsubscribe(subscriptionEndpoint); + std::this_thread::sleep_for(200ms); // get a few subscription updates + int subscriptionUpdatesAfterUnsubscribe = subscriptionUpdatesReceived; + std::this_thread::sleep_for(5s); // get a few subscription updates + expect(eq(subscriptionUpdatesReceived.load(), subscriptionUpdatesAfterUnsubscribe)); + } + std::println("# test rda3 subscribe - finished"); + }; + + "CmwLightClientGetDev"_test = [] { + std::println("# Test rda3 get/set - starting"); + using namespace opencmw; + using namespace std::literals; + const std::string DEVICE_NAME = "GSITemplateDevice"; + const std::string STATUS_PROPERTY = "/GSITemplateDevice/Status"; + const std::string SETTING_PROPERTY = "/GSITemplateDevice/Setting"; + const std::string ACQUISITION_PROPERTY = "/GSITemplateDevice/Acquisition"; + const std::string SELECTOR = "FAIR.SELECTOR.C=3:S=1:P=1:T=300"; + + const char* env_nameserver = std::getenv("CMW_NAMESERVER"); + if (env_nameserver == nullptr) { + std::println("skipping BasicCmwLight example test as it relies on the availability of network infrastructure. Define CMW_NAMESERVER environment variable to run this test."); + return; // skip test + } + std::string nameserver{env_nameserver}; + const zmq::Context zctx{}; + std::vector> clients; + auto cmwlightClient = std::make_unique(zctx, nameserver, 100ms, "testclient"); + std::string deviceHost{}; // "tcp://vmla017:36725" }; + if (const char* env_cmw_host = std::getenv("CMW_DEVICE_HOST"); env_cmw_host != nullptr) { + deviceHost = std::string{env_cmw_host}; + cmwlightClient->nameserverClient().addStaticLookup(DEVICE_NAME, std::string{env_cmw_host}); + } else { + while (!deviceHost.empty()) { + deviceHost = cmwlightClient->nameserverClient().lookup(DEVICE_NAME).value_or(""); + } + } + clients.emplace_back(std::move(cmwlightClient)); + client::ClientContext clientContext{std::move(clients)}; + // send some requests + { // non-multiplexed get request + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(STATUS_PROPERTY).build(); + std::atomic getReceived{0}; + clientContext.get(endpoint, [&getReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "get should have succeeded"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 13_i) << "wrong number of field in subscription update"; + ++getReceived; + } + }); + expect(waitFor(getReceived, 1)) << "timed-out waiting for get response"; + } + { // multiplexed get request + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(ACQUISITION_PROPERTY).addQueryParameter("ctx", SELECTOR).build(); + std::atomic getReceived{0}; + clientContext.get(endpoint, [&getReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "get should have succeeded"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 12_i) << "wrong number of field in subscription update"; + ++getReceived; + } + }); + expect(waitFor(getReceived, 1)) << "timed-out waiting for get response"; + } + { // multiplexed set request + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(SETTING_PROPERTY).addQueryParameter("ctx", SELECTOR).build(); + std::atomic setReceived{0}; + TestSettings testSettings{}; + IoBuffer data{}; + opencmw::serialise(data, testSettings); + clientContext.set( + endpoint, + [&setReceived](const mdp::Message& message) { + if (!message.error.empty()) { + std::println("Error: {}", message.error); + expect(false) << "set should have succeeded"; + } else { + ++setReceived; + } + }, + std::move(data)); + expect(waitFor(setReceived, 1)) << "timed-out waiting for set response"; + } + { // error-case: multiplexed get request with missing selector + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(SETTING_PROPERTY).build(); + std::atomic getError{0}; + clientContext.get(endpoint, [&getError](const mdp::Message& message) { + if (!message.error.empty()) { + expect(message.error.contains("Access point 'GSITemplateDevice/Setting' needs a selector")); + ++getError; + } else { + expect(false) << "get should have failed"; + } + }); + expect(waitFor(getError, 1)) << "timed-out waiting for get error"; + } + { // error-case: non-existent device property + auto endpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path("GSITemplateDevice/NON_EXISTENT_PROPERTY").build(); + std::atomic getError{0}; + clientContext.get(endpoint, [&getError](const mdp::Message& message) { + if (!message.error.empty()) { + std::println("Error: {}", message.error); + expect(message.error.contains("Error in request: UNKNOWN_PROPERTY: Device 'GSITemplateDevice' has no property named 'NON_EXISTENT_PROPERTY'")); + ++getError; + } else { + expect(false) << "get should have failed"; + } + }); + expect(waitFor(getError, 1)) << "timed-out waiting for get error"; + } + std::println("# Test rda3 get/set - finished"); + }; + + "CmwLightClientSubscribeDev"_test = [] { + std::println("# test rda3 subscribe - starting"); + using namespace opencmw; + using namespace std::literals; + const std::string DEVICE_NAME = "GSITemplateDevice"; + const std::string STATUS_PROPERTY = "/GSITemplateDevice/Status"; + const std::string ACQUISITION_PROPERTY = "/GSITemplateDevice/Acquisition"; + const std::string SELECTOR = "FAIR.SELECTOR.P=1"; + + const char* env_nameserver = std::getenv("CMW_NAMESERVER"); + if (env_nameserver == nullptr) { + std::println("skipping BasicCmwLight example test as it relies on the availability of network infrastructure. Define CMW_NAMESERVER environment variable to run this test."); + return; // skip test + } + std::string nameserver{env_nameserver}; + const zmq::Context zctx{}; + std::vector> clients; + auto cmwlightClient = std::make_unique(zctx, nameserver, 100ms, "testclient"); + std::string deviceHost{}; // "tcp://vmla017:36725" }; + if (const char* env_cmw_host = std::getenv("CMW_DEVICE_HOST"); env_cmw_host != nullptr) { + deviceHost = std::string{env_cmw_host}; + cmwlightClient->nameserverClient().addStaticLookup(DEVICE_NAME, std::string{env_cmw_host}); + } else { + while (!deviceHost.empty()) { + deviceHost = cmwlightClient->nameserverClient().lookup(DEVICE_NAME).value_or(""); + } + } + clients.emplace_back(std::move(cmwlightClient)); + client::ClientContext clientContext{std::move(clients)}; + { // Subscribe non-multiplexed + std::atomic subscriptionUpdatesReceived{0}; + auto subscriptionEndpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(STATUS_PROPERTY).build(); + clientContext.subscribe(subscriptionEndpoint, [&subscriptionUpdatesReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "subscription should not notify exceptions"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 13_i) << "wrong number of field in subscription update"; + ++subscriptionUpdatesReceived; + } + }); + expect(waitFor(subscriptionUpdatesReceived, 2, 5s)) << "timed-out waiting for subscription updates"; // property gets notified every 2 seconds + // Check that subscription updates stop after unsubscribing + clientContext.unsubscribe(subscriptionEndpoint); + std::this_thread::sleep_for(200ms); // get a few subscription updates + int subscriptionUpdatesAfterUnsubscribe = subscriptionUpdatesReceived; + std::this_thread::sleep_for(5s); // get a few subscription updates + expect(eq(subscriptionUpdatesReceived.load(), subscriptionUpdatesAfterUnsubscribe)); + } + { // error case: subscribe to multiplexed without selector + std::atomic subscriptionUpdateError{0}; + auto subscriptionEndpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(ACQUISITION_PROPERTY).build(); + clientContext.subscribe(subscriptionEndpoint, [&subscriptionUpdateError](const mdp::Message& message) { + if (!message.error.empty()) { + expect(message.error.contains("Access point 'GSITemplateDevice/Acquisition' needs a selector")); + ++subscriptionUpdateError; + } else { + expect(false) << "invalid subscription should not notify updates"; + } + }); + expect(waitFor(subscriptionUpdateError, 1, 1s)) << "timed-out waiting for subscription error"; + } + { // subscribe to multiplexed property + std::atomic subscriptionUpdatesReceived{0}; + auto subscriptionEndpoint = URI<>::factory(URI(deviceHost)).scheme("rda3tcp").path(ACQUISITION_PROPERTY).addQueryParameter("ctx", SELECTOR).build(); + std::println("Subscribing to acquisition property with url {}", subscriptionEndpoint); + clientContext.subscribe(subscriptionEndpoint, [&subscriptionUpdatesReceived](const mdp::Message& message) { + if (!message.error.empty()) { + expect(false) << "subscription should not notify exceptions"; + } else { + IoBuffer buffer(message.data); + majordomo::Empty empty{}; + const auto deserialiserInfo = deserialise(buffer, empty); // deserialising into an empty struct to get field information + std::println("Deserialised subscription reply:\n fields: {}\n fieldTypes: {}", deserialiserInfo.additionalFields | std::views::keys, deserialiserInfo.additionalFields | std::views::values); + expect(deserialiserInfo.additionalFields.size() == 12_i) << "wrong number of fields in subscription update"; + ++subscriptionUpdatesReceived; + } + }); + expect(waitFor(subscriptionUpdatesReceived, 2, 5s)) << "timed-out waiting for subscription updates"; // property gets notified every 2 seconds + // Check that subscription updates stop after unsubscribing + clientContext.unsubscribe(subscriptionEndpoint); + std::this_thread::sleep_for(200ms); // get a few subscription updates + int subscriptionUpdatesAfterUnsubscribe = subscriptionUpdatesReceived; + std::this_thread::sleep_for(5s); // get a few subscription updates + expect(eq(subscriptionUpdatesReceived.load(), subscriptionUpdatesAfterUnsubscribe)); + } + std::println("# test rda3 subscribe - finished"); + }; +}; + +int main() { /* tests are statically executed */ } diff --git a/blocklib/opencmw/test/qa_OpenCmwSource.cpp b/blocklib/opencmw/test/qa_OpenCmwSource.cpp new file mode 100644 index 00000000..7a0bbbf8 --- /dev/null +++ b/blocklib/opencmw/test/qa_OpenCmwSource.cpp @@ -0,0 +1,100 @@ +#include +#include +#include + +#include +#include +#include +#include + +#include + +#include + +#include "ProcessHelper.hpp" + +const boost::ut::suite OpenCmwSourceTests = [] { + using namespace boost::ut; + using namespace std::literals; + using namespace gr; + using namespace gr::testing; + + "OpenCmwSourceBlockDev"_test = [](bool verbose = true) { + /* This unit-test depends on the availability of internal infrastructure and will be skipped if the required environment variables are not set up */ + std::println("# CmwSourceBlock DEV - started"); + + const char* env_nameserver = std::getenv("CMW_NAMESERVER"); + if (env_nameserver == nullptr) { + std::println("skipping BasicCmwLight example test as it relies on the availability of network infrastructure. Define CMW_NAMESERVER environment variable to run this test."); + return; // skip test + } + + std::string DEVICE_NAME = "GSITemplateDevice"; + std::string deviceHost{""}; // by default, perform name lookup on the nameserver, explicitly setting this to localhost might be needed in case ssh tunnels are required + if (const char* env_cmw_host = std::getenv("CMW_DEVICE_HOST"); env_cmw_host != nullptr) { + deviceHost = std::string{env_cmw_host}; + } + + Graph testGraph; + auto& src = testGraph.emplaceBlock({ + {"url", "rda3://" + deviceHost + "/GSITemplateDevice/Acquisition?ctx=FAIR.SELECTOR.ALL"}, + {"verbose_console", verbose}, + }); + auto& extractFromMapBlock = testGraph.emplaceBlock>({{"fieldname", "voltage"}}); + auto& sink = testGraph.emplaceBlock>({{"name", "TagSink1"}}); + expect(eq(ConnectionResult::SUCCESS, testGraph.connect<"out">(src).template to<"in">(extractFromMapBlock))); + expect(eq(ConnectionResult::SUCCESS, testGraph.connect<"out">(extractFromMapBlock).template to<"in">(sink))); + + scheduler::Simple sched{}; + std::ignore = sched.exchange(std::move(testGraph)); + MsgPortOut _toScheduler; + expect(_toScheduler.connect(sched.msgIn) == ConnectionResult::SUCCESS) << fatal; + expect(sched.changeStateTo(lifecycle::State::INITIALISED).has_value()); + expect(sched.changeStateTo(lifecycle::State::RUNNING).has_value()); + std::this_thread::sleep_for(5s); + std::println("stopping scheduler"); + gr::sendMessage(_toScheduler, sched.unique_name, block::property::kLifeCycleState, {{"state", std::string(magic_enum::enum_name(lifecycle::State::REQUESTED_STOP))}}, "test"); + + expect(sink._nSamplesProduced == 4_i); + std::println("sink received samples: {}", sink._samples); + + std::println("# CmwSourceBlock - finished"); + }; + + "OpenCmwSourceBlockLocal"_test = [](bool verbose = true) { + std::println("# CmwSourceBlock local mock rda3 server - started"); + + Process rda3Server({"java", "-jar", "./DemoRda3Server.jar", "-cfg", "--cmw.rda3.transport.server.port=12345", "-sleep", "1400"}); + std::this_thread::sleep_for(10ms); // some time for rda3 server startup + + std::string DEVICE_NAME = "GSITemplateDevice"; + std::string deviceHost{"rda3://localhost:12345"}; + + Graph testGraph; + auto& src = testGraph.emplaceBlock({ + {"url", deviceHost + "/GSITemplateDevice/Acquisition?ctx=FAIR.SELECTOR.ALL"}, + {"verbose_console", verbose}, + }); + auto& extractFromMapBlock = testGraph.emplaceBlock>({{"fieldname", "voltage"}}); + auto& sink = testGraph.emplaceBlock>({{"name", "TagSink1"}}); + expect(eq(ConnectionResult::SUCCESS, testGraph.connect<"out">(src).template to<"in">(extractFromMapBlock))); + expect(eq(ConnectionResult::SUCCESS, testGraph.connect<"out">(extractFromMapBlock).template to<"in">(sink))); + + scheduler::Simple sched{}; + std::ignore = sched.exchange(std::move(testGraph)); + MsgPortOut _toScheduler; + expect(_toScheduler.connect(sched.msgIn) == ConnectionResult::SUCCESS) << fatal; + expect(sched.changeStateTo(lifecycle::State::INITIALISED).has_value()); + expect(sched.changeStateTo(lifecycle::State::RUNNING).has_value()); + std::this_thread::sleep_for(5s); + std::println("stopping scheduler"); + gr::sendMessage(_toScheduler, sched.unique_name, block::property::kLifeCycleState, {{"state", std::string(magic_enum::enum_name(lifecycle::State::REQUESTED_STOP))}}, "test"); + + expect(sink._nSamplesProduced == 4_i); + std::println("sink received samples: {}", sink._samples); + + std::println("# CmwSourceBlock - finished"); + }; +}; + +int main() { /* tests are statically executed */ }