Repurpose into VIS clientneedlefish_13.93.0needlefish/13.93.013.93.0

Repurpose repository for a spiritual successor of the previous
binding.  The replacement is a daemon that demonstrates servicing
HVAC actuators from the VSS schema via VIS signals from KUKSA.val.
Currently the connection to KUKSA.val is websocket based using the
boost::asio framework, but the plan is to migrate to grpc as that
becomes more robust in KUKSA.val.

Notable changes:
- New code is completely C++, partly to leverage using Boost, but
  also to futureproof future work with grpc.
- Switch from CMake to meson for ease of development and some
  degree of futureproofing.
- Use with systemd is assumed; behavior follows the systemd
  daemon guidelines barring the use of journald logging prefixes,
  which may be addressed with future work.  A systemd unit is
  also installed as part of the build.
- SPDX license headers using SPDX "short identifiers" are used in
  source files rather than the full copyright headers used in the
  previous codebase.  This follows the direction that projects such
  as the Linux kernel are going in.
- The JSON configuration file for the LED control files for the
  demo platform has been migrated to a INI format configuration
  file matching what has been done for the VIS client configuration
  in other recent work.

Bug-AGL: SPEC-4409

Signed-off-by: Scott Murray <scott.murray@konsulko.com>
Change-Id: Ic2061bca9670b1e461d6f1e6591471e257fff5b9

1 files changed, 374 insertions, 0 deletions

diff --git a/src/vis-session.cpp b/src/vis-session.cpp
new file mode 100644
index 0000000..880e3ae
--- /dev/null
+++ b/src/vis-session.cpp
@@ -0,0 +1,374 @@
+// SPDX-License-Identifier: Apache-2.0
+
+#include "vis-session.hpp"
+#include <iostream>
+#include <sstream>
+#include <thread>
+
+
+// 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<int>(::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 &timestamp)
+{
+	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_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;
+}
+