// SPDX-License-Identifier: Apache-2.0 #include "vis-session.hpp" #include #include #include // Logging helper static void log_error(beast::error_code error, char const* what) { std::cerr << what << " error: " << error.message() << std::endl; } // Resolver and socket require an io_context VisSession::VisSession(const VisConfig &config, net::io_context& ioc, ssl::context& ctx) : m_config(config), m_resolver(net::make_strand(ioc)), m_ws(net::make_strand(ioc), ctx) { } // Start the asynchronous operation void VisSession::run() { if (!m_config.valid()) { return; } // Start by resolving hostname m_resolver.async_resolve(m_config.hostname(), std::to_string(m_config.port()), beast::bind_front_handler(&VisSession::on_resolve, shared_from_this())); } void VisSession::on_resolve(beast::error_code error, tcp::resolver::results_type results) { if(error) { log_error(error, "resolve"); return; } // Set a timeout on the connect operation beast::get_lowest_layer(m_ws).expires_after(std::chrono::seconds(30)); // Connect to resolved address if (m_config.verbose()) std::cout << "Connecting" << std::endl; m_results = results; connect(); } void VisSession::connect() { beast::get_lowest_layer(m_ws).async_connect(m_results, beast::bind_front_handler(&VisSession::on_connect, shared_from_this())); } void VisSession::on_connect(beast::error_code error, tcp::resolver::results_type::endpoint_type endpoint) { if(error) { // The server can take a while to be ready to accept connections, // so keep retrying until we hit the timeout. if (error == net::error::timed_out) { log_error(error, "connect"); return; } // Delay 500 ms before retrying std::this_thread::sleep_for(std::chrono::milliseconds(500)); if (m_config.verbose()) std::cout << "Connecting" << std::endl; connect(); return; } if (m_config.verbose()) std::cout << "Connected" << std::endl; // Set handshake timeout beast::get_lowest_layer(m_ws).expires_after(std::chrono::seconds(30)); // Set SNI Hostname (many hosts need this to handshake successfully) if(!SSL_set_tlsext_host_name(m_ws.next_layer().native_handle(), m_config.hostname().c_str())) { error = beast::error_code(static_cast(::ERR_get_error()), net::error::get_ssl_category()); log_error(error, "connect"); return; } // Update the hostname. This will provide the value of the // Host HTTP header during the WebSocket handshake. // See https://tools.ietf.org/html/rfc7230#section-5.4 m_hostname = m_config.hostname() + ':' + std::to_string(endpoint.port()); if (m_config.verbose()) std::cout << "Negotiating SSL handshake" << std::endl; // Perform the SSL handshake m_ws.next_layer().async_handshake(ssl::stream_base::client, beast::bind_front_handler(&VisSession::on_ssl_handshake, shared_from_this())); } void VisSession::on_ssl_handshake(beast::error_code error) { if(error) { log_error(error, "SSL handshake"); return; } // Turn off the timeout on the tcp_stream, because // the websocket stream has its own timeout system. beast::get_lowest_layer(m_ws).expires_never(); // NOTE: Explicitly not setting websocket stream timeout here, // as the client is long-running. if (m_config.verbose()) std::cout << "Negotiating WSS handshake" << std::endl; // Perform handshake m_ws.async_handshake(m_hostname, "/", beast::bind_front_handler(&VisSession::on_handshake, shared_from_this())); } void VisSession::on_handshake(beast::error_code error) { if(error) { log_error(error, "WSS handshake"); return; } if (m_config.verbose()) std::cout << "Authorizing" << std::endl; // Authorize json req; req["requestId"] = std::to_string(m_requestid++); req["action"]= "authorize"; req["tokens"] = m_config.authToken(); m_ws.async_write(net::buffer(req.dump(4)), beast::bind_front_handler(&VisSession::on_authorize, shared_from_this())); } void VisSession::on_authorize(beast::error_code error, std::size_t bytes_transferred) { boost::ignore_unused(bytes_transferred); if(error) { log_error(error, "authorize"); return; } // Read response m_ws.async_read(m_buffer, beast::bind_front_handler(&VisSession::on_read, shared_from_this())); } // NOTE: Placeholder for now void VisSession::on_write(beast::error_code error, std::size_t bytes_transferred) { boost::ignore_unused(bytes_transferred); if(error) { log_error(error, "write"); return; } // Do nothing... } void VisSession::on_read(beast::error_code error, std::size_t bytes_transferred) { boost::ignore_unused(bytes_transferred); if(error) { log_error(error, "read"); return; } // Handle message std::string s = beast::buffers_to_string(m_buffer.data()); json response = json::parse(s, nullptr, false); if (!response.is_discarded()) { handle_message(response); } else { std::cerr << "json::parse failed? got " << s << std::endl; } m_buffer.consume(m_buffer.size()); // Read next message m_ws.async_read(m_buffer, beast::bind_front_handler(&VisSession::on_read, shared_from_this())); } void VisSession::get(const std::string &path) { if (!m_config.valid()) { return; } json req; req["requestId"] = std::to_string(m_requestid++); req["action"] = "get"; req["path"] = path; req["tokens"] = m_config.authToken(); m_ws.write(net::buffer(req.dump(4))); } void VisSession::set(const std::string &path, const std::string &value) { if (!m_config.valid()) { return; } json req; req["requestId"] = std::to_string(m_requestid++); req["action"] = "set"; req["path"] = path; req["value"] = value; req["tokens"] = m_config.authToken(); m_ws.write(net::buffer(req.dump(4))); } void VisSession::subscribe(const std::string &path) { if (!m_config.valid()) { return; } json req; req["requestId"] = std::to_string(m_requestid++); req["action"] = "subscribe"; req["path"] = path; req["tokens"] = m_config.authToken(); m_ws.write(net::buffer(req.dump(4))); } bool VisSession::parseData(const json &message, std::string &path, std::string &value, std::string ×tamp) { if (message.contains("error")) { std::string error = message["error"]; return false; } if (!(message.contains("data") && message["data"].is_object())) { std::cerr << "Malformed message (data missing)" << std::endl; return false; } auto data = message["data"]; if (!(data.contains("path") && data["path"].is_string())) { std::cerr << "Malformed message (path missing)" << std::endl; return false; } path = data["path"]; // Convert '/' to '.' in paths to ensure consistency for clients std::replace(path.begin(), path.end(), '/', '.'); if (!(data.contains("dp") && data["dp"].is_object())) { std::cerr << "Malformed message (datapoint missing)" << std::endl; return false; } auto dp = data["dp"]; if (!dp.contains("value")) { std::cerr << "Malformed message (value missing)" << std::endl; return false; } else if (dp["value"].is_string()) { value = dp["value"]; } else if (dp["value"].is_number_float()) { double num = dp["value"]; value = std::to_string(num); } else if (dp["value"].is_number_unsigned()) { unsigned int num = dp["value"]; value = std::to_string(num); } else if (dp["value"].is_number_integer()) { int num = dp["value"]; value = std::to_string(num); } else if (dp["value"].is_boolean()) { value = dp["value"] ? "true" : "false"; } else { std::cerr << "Malformed message (unsupported value type)" << std::endl; return false; } if (!(dp.contains("ts") && dp["ts"].is_string())) { std::cerr << "Malformed message (timestamp missing)" << std::endl; return false; } timestamp = dp["ts"]; return true; } void VisSession::handle_message(const json &message) { if (m_config.verbose() > 1) std::cout << "VisSession::handle_message: enter, message = " << to_string(message) << std::endl; if (!message.contains("action")) { std::cerr << "Received unknown message (no action), discarding" << std::endl; return; } std::string action = message["action"]; if (action == "authorize") { if (message.contains("error")) { std::string error = "unknown"; if (message["error"].is_object() && message["error"].contains("message")) error = message["error"]["message"]; std::cerr << "VIS authorization failed: " << error << std::endl; } else { if (m_config.verbose() > 1) std::cout << "authorized" << std::endl; handle_authorized_response(); } } else if (action == "subscribe") { if (message.contains("error")) { std::string error = "unknown"; if (message["error"].is_object() && message["error"].contains("message")) error = message["error"]["message"]; std::cerr << "VIS subscription failed: " << error << std::endl; } } else if (action == "get") { if (message.contains("error")) { std::string error = "unknown"; if (message["error"].is_object() && message["error"].contains("message")) error = message["error"]["message"]; std::cerr << "VIS get failed: " << error << std::endl; } else { std::string path, value, ts; if (parseData(message, path, value, ts)) { if (m_config.verbose() > 1) std::cout << "VisSession::handle_message: got response " << path << " = " << value << std::endl; handle_get_response(path, value, ts); } } } else if (action == "set") { if (message.contains("error")) { std::string error = "unknown"; if (message["error"].is_object() && message["error"].contains("message")) error = message["error"]["message"]; std::cerr << "VIS set failed: " << error; } } else if (action == "subscription") { std::string path, value, ts; if (parseData(message, path, value, ts)) { if (m_config.verbose() > 1) std::cout << "VisSession::handle_message: got notification " << path << " = " << value << std::endl; handle_notification(path, value, ts); } } else { std::cerr << "unhandled VIS response of type: " << action; } if (m_config.verbose() > 1) std::cout << "VisSession::handle_message: exit" << std::endl; }