7 files changed, 846 insertions, 434 deletions

diff --git a/vehicle-signals/QtKuksaClient.cpp b/vehicle-signals/QtKuksaClient.cpp
new file mode 100644
index 0000000..901459d
--- /dev/null
+++ b/vehicle-signals/QtKuksaClient.cpp
@@ -0,0 +1,421 @@
+/*
+ * Copyright (C) 2023 Konsulko Group
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <QDebug>
+#include <QSettings>
+#include <QUrl>
+#include <QFile>
+#include <QtConcurrent>
+#include <mutex>
+#include <chrono>
+
+#include "QtKuksaClient.h"
+
+using grpc::Channel;
+using grpc::ClientContext;
+using grpc::ClientReader;
+using grpc::Status;
+
+class QtKuksaClient::SubscribeReader : public grpc::ClientReadReactor<SubscribeResponse> {
+public:
+	SubscribeReader(VAL::Stub *stub,
+			QtKuksaClient *client,
+			VehicleSignalsConfig &config,
+			const QMap<QString, bool> &s,
+			const SubscribeRequest *request):
+		client_(client),
+		config_(config),
+		signals_(s) {
+		QString token = config_.authToken();
+		if (!token.isEmpty()) {
+			token.prepend("Bearer ");
+			context_.AddMetadata(std::string("authorization"), token.toStdString());
+		}
+		stub->async()->Subscribe(&context_, request, this);
+		StartRead(&response_);
+		StartCall();
+	}
+	void OnReadDone(bool ok) override {
+		std::unique_lock<std::mutex> lock(mutex_);
+		if (ok) {
+			if (client_)
+				client_->handleSubscribeResponse(&response_);
+			StartRead(&response_);
+		}
+	}
+	void OnDone(const Status& s) override {
+		status_ = s;
+		if (client_) {
+			if (config_.verbose() > 1)
+				qDebug() << "QtKuksaClient::subscribe::Reader done";
+			client_->handleSubscribeDone(signals_, status_);
+		}
+
+		// gRPC engine is done with us, safe to self-delete
+		delete this;
+	}
+
+private:
+	QtKuksaClient *client_;
+	VehicleSignalsConfig config_;
+	QMap<QString, bool> signals_;
+
+	ClientContext context_;
+	SubscribeResponse response_;
+	std::mutex mutex_;
+	Status status_;
+};
+
+QtKuksaClient::QtKuksaClient(const std::shared_ptr< ::grpc::ChannelInterface>& channel,
+			     const VehicleSignalsConfig &config,
+			     QObject *parent) :
+	QObject(parent),
+	m_channel(channel),
+	m_config(config),
+	m_connected(false)
+{
+        m_stub = VAL::NewStub(channel);
+
+}
+
+void QtKuksaClient::connect()
+{
+	// Check for connection in another thread
+	QFuture<void> future = QtConcurrent::run(this, &QtKuksaClient::waitForConnected);
+}
+
+void QtKuksaClient::get(const QString &path, const bool actuator)
+{
+	m_connected_mutex.lock();
+	if (!m_connected) {
+		m_connected_mutex.unlock();
+		return;
+	}
+	m_connected_mutex.unlock();
+	
+	ClientContext *context = new ClientContext();
+	if (!context) {
+		handleCriticalFailure("Could not create ClientContext");
+		return;
+	}
+	QString token = m_config.authToken();
+	if (!token.isEmpty()) {
+		token.prepend("Bearer ");
+		context->AddMetadata(std::string("authorization"), token.toStdString());
+	}
+
+	GetRequest request;
+	auto entry = request.add_entries();
+	entry->set_path(path.toStdString());
+	entry->add_fields(Field::FIELD_PATH);	
+	if (actuator)
+		entry->add_fields(Field::FIELD_ACTUATOR_TARGET);
+	else
+		entry->add_fields(Field::FIELD_VALUE);
+
+	GetResponse *response = new GetResponse();
+	if (!response) {
+		handleCriticalFailure("Could not create GetResponse");
+		return;
+	}
+
+	// NOTE: Using ClientUnaryReactor instead of the shortcut method
+	//       would allow getting detailed errors.
+	m_stub->async()->Get(context, &request, response,
+			     [this, context, response](Status s) {
+				     if (s.ok())
+					     handleGetResponse(response);
+				     delete response;
+				     delete context;
+			     });
+}
+
+// Since a set request needs a Datapoint with the appropriate type value,
+// checking the signal metadata to get the type would be a requirement for
+// a generic set call that takes a string as argument.  For now, assume
+// that set with a string is specifically for a signal of string type.
+
+void QtKuksaClient::set(const QString &path, const QString &value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_string(value.toStdString());
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::set(const QString &path, const qint32 value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_int32(value);
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::set(const QString &path, const qint64 value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_int64(value);
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::set(const QString &path, const quint32 value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_uint32(value);
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::set(const QString &path, const quint64 value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_uint64(value);
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::set(const QString &path, const float value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_float_(value);
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::set(const QString &path, const double value, const bool actuator)
+{
+	Datapoint dp;
+	dp.set_double_(value);
+	set(path, dp, actuator);
+}
+
+void QtKuksaClient::subscribe(const QString &path, const bool actuator)
+{
+	m_connected_mutex.lock();
+	if (!m_connected) {
+		m_connected_mutex.unlock();
+		return;
+	}
+	m_connected_mutex.unlock();
+
+	QMap<QString, bool> s;
+	s[path] = actuator;
+	subscribe(s);
+}
+
+void QtKuksaClient::subscribe(const QMap<QString, bool> &signals_)
+{
+	m_connected_mutex.lock();
+	if (!m_connected) {
+		m_connected_mutex.unlock();
+		return;
+	}
+	m_connected_mutex.unlock();
+
+	SubscribeRequest *request = new SubscribeRequest();
+	if (!request) {
+		handleCriticalFailure("Could not create SubscribeRequest");
+		return;
+	}
+
+	auto it = signals_.constBegin();
+	while (it != signals_.constEnd()) {
+		if (m_config.verbose() > 1)
+			qDebug() << "QtKuksaClient::subscribe: adding " << it.key() << ", actuator " << it.value();
+		auto entry = request->add_entries();
+		entry->set_path(it.key().toStdString());
+		entry->add_fields(Field::FIELD_PATH);	
+		if (it.value())
+			entry->add_fields(Field::FIELD_ACTUATOR_TARGET);
+		else
+			entry->add_fields(Field::FIELD_VALUE);
+		++it;
+	}
+
+	SubscribeReader *reader = new SubscribeReader(m_stub.get(), this, m_config, signals_, request);
+	if (!reader)
+		handleCriticalFailure("Could not create SubscribeReader");
+
+	delete request;
+}
+
+// Private
+
+void QtKuksaClient::waitForConnected()
+{
+	while (!m_channel->WaitForConnected(std::chrono::system_clock::now() +
+					  std::chrono::milliseconds(500)));
+
+	m_connected_mutex.lock();
+	m_connected = true;
+	m_connected_mutex.unlock();
+
+	qInfo() << "Databroker gRPC channel ready";
+	emit connected();
+}
+
+void QtKuksaClient::set(const QString &path, const Datapoint &dp, const bool actuator)
+{
+	m_connected_mutex.lock();
+	if (!m_connected) {
+		m_connected_mutex.unlock();
+		return;
+	}
+	m_connected_mutex.unlock();
+
+	ClientContext *context = new ClientContext();
+	if (!context) {
+		handleCriticalFailure("Could not create ClientContext");
+		return;
+	}
+	QString token = m_config.authToken();
+	if (!token.isEmpty()) {
+		token.prepend("Bearer ");
+		context->AddMetadata(std::string("authorization"), token.toStdString());
+	}
+
+	SetRequest request;
+	auto update = request.add_updates();
+	auto entry = update->mutable_entry();
+	entry->set_path(path.toStdString());
+	if (actuator) {
+		auto target = entry->mutable_actuator_target();
+		*target = dp;
+		update->add_fields(Field::FIELD_ACTUATOR_TARGET);
+	} else {
+		auto value = entry->mutable_value();
+		*value = dp;
+		update->add_fields(Field::FIELD_VALUE);		
+	}
+
+	SetResponse *response = new SetResponse();
+	if (!response) {
+		handleCriticalFailure("Could not create SetResponse");
+		delete context;
+		return;
+	}
+
+	// NOTE: Using ClientUnaryReactor instead of the shortcut method
+	//       would allow getting detailed errors.
+	m_stub->async()->Set(context, &request, response,
+				       [this, context, response](Status s) {
+					       if (s.ok())
+						       handleSetResponse(response);
+					       delete response;
+					       delete context;
+				       });
+}
+
+void QtKuksaClient::handleGetResponse(const GetResponse *response)
+{
+	if (!(response && response->entries_size()))
+		return;
+
+	for (auto it = response->entries().begin(); it != response->entries().end(); ++it) {
+		// We expect paths in the response entries
+		if (!it->path().size())
+			continue;
+		Datapoint dp;
+		if (it->has_actuator_target())
+			dp = it->actuator_target();
+		else
+			dp = it->value();
+
+		QString path = QString::fromStdString(it->path());
+		QString value;
+		QString timestamp;		
+		if (convertDatapoint(dp, value, timestamp)) {
+			if (m_config.verbose() > 1)
+				qDebug() << "QtKuksaClient::handleGetResponse: path = " << path << ", value = " << value;
+
+			emit getResponse(path, value, timestamp);
+		}
+	}
+}
+
+void QtKuksaClient::handleSetResponse(const SetResponse *response)
+{
+	if (!(response && response->errors_size()))
+		return;
+
+	for (auto it = response->errors().begin(); it != response->errors().end(); ++it) {
+		QString path = QString::fromStdString(it->path());
+		QString error = QString::fromStdString(it->error().reason());
+		emit setResponse(path, error);
+	}
+
+}
+
+void QtKuksaClient::handleSubscribeResponse(const SubscribeResponse *response)
+{
+	if (!(response && response->updates_size()))
+		return;
+
+	for (auto it = response->updates().begin(); it != response->updates().end(); ++it) {
+		// We expect entries that have paths in the response
+		if (!(it->has_entry() && it->entry().path().size()))
+			continue;
+
+		auto entry = it->entry();
+		QString path = QString::fromStdString(entry.path());
+
+		Datapoint dp;
+		if (entry.has_actuator_target())
+			dp = entry.actuator_target();
+		else
+			dp = entry.value();
+
+		QString value;
+		QString timestamp;		
+		if (convertDatapoint(dp, value, timestamp)) {
+			if (m_config.verbose() > 1)
+				qDebug() << "QtKuksaClient::handleSubscribeResponse: path = " << path << ", value = " << value;
+	
+			emit subscribeResponse(path, value, timestamp);
+		}
+	}
+}
+
+void QtKuksaClient::handleSubscribeDone(const QMap<QString, bool> &signals_, const Status &status)
+{
+	qInfo() << "QtKuksaClient::handleSubscribeDone: enter";
+	if (m_config.verbose() > 1)
+		qDebug() << "Subscribe status = " << status.error_code() <<
+			" (" << QString::fromStdString(status.error_message()) << ")";
+
+	emit subscribeDone(signals_, status.error_code() == grpc::CANCELLED);
+}
+
+bool QtKuksaClient::convertDatapoint(const Datapoint &dp, QString &value, QString &timestamp)
+{
+	// NOTE: Currently no array type support
+
+	if (dp.has_string())
+		value = QString::fromStdString(dp.string());
+	else if (dp.has_int32())
+		value = QString::number(dp.int32());
+	else if (dp.has_int64())
+		value = QString::number(dp.int64());
+	else if (dp.has_uint32())
+		value = QString::number(dp.uint32());
+	else if (dp.has_uint64())
+		value = QString::number(dp.uint64());
+	else if (dp.has_float_())
+		value = QString::number(dp.float_(), 'f', 6);
+	else if (dp.has_double_())
+		value = QString::number(dp.double_(), 'f', 6);
+	else if (dp.has_bool_())
+		value = dp.bool_() ? "true" : "false";
+	else {
+		if (m_config.verbose())
+			qWarning() << "Malformed response (unsupported value type)";
+		return false;
+	}
+
+	timestamp = QString::number(dp.timestamp().seconds()) + "." + QString::number(dp.timestamp().nanos());
+	return true;
+}
+
+void QtKuksaClient::handleCriticalFailure(const QString &error)
+{
+	if (error.size())
+		qCritical() << error;
+}
diff --git a/vehicle-signals/QtKuksaClient.h b/vehicle-signals/QtKuksaClient.h
new file mode 100644
index 0000000..1f24e09
--- /dev/null
+++ b/vehicle-signals/QtKuksaClient.h
@@ -0,0 +1,86 @@
+/*
+ * Copyright (C) 2023 Konsulko Group
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef QT_KUKSA_CLIENT_H
+#define QT_KUKSA_CLIENT_H
+
+#include <QObject>
+#include <QMap>
+#include <QMutex>
+#include <grpcpp/grpcpp.h>
+#include "kuksa/val/v1/val.grpc.pb.h"
+
+// Just pull in the whole namespace since Datapoint contains a lot of
+// definitions that may potentially be needed.
+using namespace kuksa::val::v1;
+
+using grpc::Status;
+
+#include "VehicleSignalsConfig.h"
+
+// KUKSA.val databroker gRPC API Qt client class
+
+class QtKuksaClient : public QObject
+{
+	Q_OBJECT
+
+public:
+	explicit QtKuksaClient(const std::shared_ptr< ::grpc::ChannelInterface>& channel,
+			       const VehicleSignalsConfig &config,
+			       QObject * parent = Q_NULLPTR);
+
+	Q_INVOKABLE void connect();
+
+	Q_INVOKABLE void get(const QString &path, const bool actuator = false);
+
+	Q_INVOKABLE void set(const QString &path, const QString &value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const qint32 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const qint64 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const quint32 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const quint64 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const float value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const double value, const bool actuator = false);
+  
+	Q_INVOKABLE void subscribe(const QString &path, const bool actuator = false);
+	Q_INVOKABLE void subscribe(const QMap<QString, bool> &signals_);
+
+signals:
+	void connected();
+	void getResponse(QString path, QString value, QString timestamp);
+        void setResponse(QString path, QString error);
+        void subscribeResponse(QString path, QString value, QString timestamp);
+        void subscribeDone(const QMap<QString, bool> &signals_, bool canceled);
+
+private:
+	class SubscribeReader;
+
+	std::shared_ptr< ::grpc::ChannelInterface> m_channel;
+	VehicleSignalsConfig m_config;
+	std::shared_ptr<VAL::Stub> m_stub;
+	bool m_connected;
+	QMutex m_connected_mutex;
+
+	void waitForConnected();
+
+	void set(const QString &path, const Datapoint &dp, const bool actuator);
+
+	void handleGetResponse(const GetResponse *response);
+
+	void handleSetResponse(const SetResponse *response);
+
+	void handleSubscribeResponse(const SubscribeResponse *response);
+
+	void handleSubscribeDone(const QMap<QString, bool> &signals_, const Status &status);
+
+	void handleCriticalFailure(const QString &error);
+	void handleCriticalFailure(const char *error) { handleCriticalFailure(QString(error)); };
+
+	void resubscribe(const QMap<QString, bool> &signals_);
+
+	bool convertDatapoint(const Datapoint &dp, QString &value, QString &timestamp);
+};
+
+#endif // QT_KUKSA_CLIENT_H
diff --git a/vehicle-signals/VehicleSignalsConfig.cpp b/vehicle-signals/VehicleSignalsConfig.cpp
new file mode 100644
index 0000000..c72c2cd
--- /dev/null
+++ b/vehicle-signals/VehicleSignalsConfig.cpp
@@ -0,0 +1,104 @@
+/*
+ * Copyright (C) 2023 Konsulko Group
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include <QDebug>
+#include <QSettings>
+#include <QUrl>
+#include <QFile>
+// ?
+//#include <QSslKey>
+//#include <QTimer>
+
+#include "VehicleSignalsConfig.h"
+
+#define DEFAULT_CLIENT_KEY_FILE  "/etc/kuksa-val/Client.key"
+#define DEFAULT_CLIENT_CERT_FILE "/etc/kuksa-val/Client.pem"
+#define DEFAULT_CA_CERT_FILE     "/etc/kuksa-val/CA.pem"
+
+VehicleSignalsConfig::VehicleSignalsConfig(const QString &hostname,
+					   const unsigned port,
+					   const QByteArray &caCert,
+					   const QString &tlsServerName,
+					   const QString &authToken) :
+	m_hostname(hostname),
+	m_port(port),
+	m_caCert(caCert),
+	m_tlsServerName(tlsServerName),
+	m_authToken(authToken),
+	m_verbose(0),
+	m_valid(true)
+{
+	// Potentially could do some certificate validation here...
+}
+
+VehicleSignalsConfig::VehicleSignalsConfig(const QString &appname)
+{
+	m_valid = false;
+
+	QSettings *pSettings = new QSettings("AGL", appname);
+	if (!pSettings)
+		return;
+
+	m_hostname = pSettings->value("kuksa-client/server", "localhost").toString();
+	if (m_hostname.isEmpty()) {
+		qCritical() << "Invalid server hostname";
+		return;
+	}
+
+	m_port = pSettings->value("kuksa-client/port", 55555).toInt();
+	if (m_port == 0) {
+		qCritical() << "Invalid server port";
+		return;
+	}
+
+	QString caCertFileName = pSettings->value("kuksa-client/ca-certificate", DEFAULT_CA_CERT_FILE).toString();
+	if (caCertFileName.isEmpty()) {
+		qCritical() << "Invalid CA certificate filename";
+		return;
+	}
+	QFile caCertFile(caCertFileName);
+	if (!caCertFile.open(QIODevice::ReadOnly)) {
+		qCritical() << "Could not open CA certificate file";
+		return;
+	}
+	QByteArray caCertData = caCertFile.readAll();
+	if (caCertData.isEmpty()) {
+		qCritical() << "Invalid CA certificate file";
+		return;
+	}
+	m_caCert = caCertData;
+
+	m_tlsServerName = pSettings->value("kuksa-client/tls-server-name", "").toString();
+
+	QString authTokenFileName = pSettings->value("kuksa-client/authorization").toString();
+	if (authTokenFileName.isEmpty()) {
+		qCritical() << "Invalid authorization token filename";
+		return;
+	}
+	QFile authTokenFile(authTokenFileName);
+	if (!authTokenFile.open(QIODevice::ReadOnly | QIODevice::Text)) {
+		qCritical() << "Could not open authorization token file";
+		return;
+	}
+	QTextStream in(&authTokenFile);
+	QString authToken = in.readLine();
+	if (authToken.isEmpty()) {
+		qCritical() << "Invalid authorization token file";
+		return;
+	}
+	m_authToken = authToken;
+
+	m_verbose = 0;
+	QString verbose = pSettings->value("kuksa-client/verbose").toString();
+	if (!verbose.isEmpty()) {
+		if (verbose == "true" || verbose == "1")
+			m_verbose = 1;
+		if (verbose == "2")
+			m_verbose = 2;
+	}
+
+	m_valid = true;
+}
diff --git a/vehicle-signals/VehicleSignalsConfig.h b/vehicle-signals/VehicleSignalsConfig.h
new file mode 100644
index 0000000..c3d52ca
--- /dev/null
+++ b/vehicle-signals/VehicleSignalsConfig.h
@@ -0,0 +1,43 @@
+/*
+ * Copyright (C) 2023 Konsulko Group
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef VEHICLE_SIGNALS_CONFIG_H
+#define VEHICLE_SIGNALS_CONFIG_H
+
+#include <QObject>
+
+// Class to read/hold VSS server configuration
+
+class VehicleSignalsConfig
+{
+public:
+        explicit VehicleSignalsConfig(const QString &hostname,
+				      const unsigned port,
+				      const QByteArray &caCert,
+				      const QString &tlsServerName,
+				      const QString &authToken);
+        explicit VehicleSignalsConfig(const QString &appname);
+        ~VehicleSignalsConfig() {};
+
+	QString hostname() { return m_hostname; };
+	unsigned port() { return m_port; };
+	QByteArray caCert() { return m_caCert; };
+	QString tlsServerName() { return m_tlsServerName; };
+	QString authToken() { return m_authToken; };
+	bool valid() { return m_valid; };
+	unsigned verbose() { return m_verbose; };
+
+private:
+	QString m_hostname;
+	unsigned m_port;
+	QByteArray m_caCert;
+	QString m_tlsServerName;
+	QString m_authToken;
+	bool m_valid = true;
+	unsigned m_verbose = 0;
+};
+
+#endif // VEHICLE_SIGNALS_CONFIG_H
diff --git a/vehicle-signals/meson.build b/vehicle-signals/meson.build
index e4c6dc0..50383e4 100644
--- a/vehicle-signals/meson.build
+++ b/vehicle-signals/meson.build
@@ -1,25 +1,69 @@
-qt5_dep = dependency('qt5', modules: ['Core', 'WebSockets'])
+cpp = meson.get_compiler('cpp')
+grpcpp_reflection_dep = cpp.find_library('grpc++_reflection')
 
-moc_files = qt5.compile_moc(headers: 'vehiclesignals.h',
+qt5_dep = dependency('qt5', modules: ['Core', 'Concurrent'])
+
+vs_dep = [
+    qt5_dep,
+    dependency('protobuf'),
+    dependency('grpc'),
+    dependency('grpc++'),
+    grpcpp_reflection_dep,
+]
+
+protoc = find_program('protoc')
+grpc_cpp = find_program('grpc_cpp_plugin')
+
+protos_base_dir = get_option('protos')
+protos_dir = protos_base_dir / 'kuksa/val/v1'
+protoc_gen = generator(protoc, \
+                       output : ['@BASENAME@.pb.cc', '@BASENAME@.pb.h'],
+                       arguments : ['-I=' + protos_base_dir,
+                         '--cpp_out=@BUILD_DIR@',
+                         '@INPUT@'])
+generated_protoc_sources = [ \
+    protoc_gen.process(protos_dir / 'types.proto', preserve_path_from : protos_base_dir),
+    protoc_gen.process(protos_dir / 'val.proto', preserve_path_from : protos_base_dir),
+]
+
+grpc_gen = generator(protoc, \
+                     output : ['@BASENAME@.grpc.pb.cc', '@BASENAME@.grpc.pb.h'],
+                     arguments : ['-I=' + protos_base_dir,
+                       '--grpc_out=@BUILD_DIR@',
+                       '--plugin=protoc-gen-grpc=' + grpc_cpp.path(),
+                       '@INPUT@'])
+generated_grpc_sources = [ \
+    grpc_gen.process(protos_dir / 'val.proto', preserve_path_from : protos_base_dir),
+]
+
+moc_files = qt5.compile_moc(headers: [ 'vehiclesignals.h', 'QtKuksaClient.h' ],
                             dependencies: qt5_dep)
 
-src = ['vehiclesignals.cpp', moc_files]
+src = [
+    'vehiclesignals.cpp',
+    'VehicleSignalsConfig.cpp',
+    'QtKuksaClient.cpp',
+    moc_files,
+    generated_protoc_sources,
+    generated_grpc_sources,
+]
+
 lib = shared_library('qtappfw-vehicle-signals',
                      sources: src,
                      version: '1.0.0',
                      soversion: '0',
-                     dependencies: qt5_dep,
+                     dependencies: vs_dep,
                      install: true)
 
-install_headers('vehiclesignals.h')
+install_headers(['vehiclesignals.h', 'VehicleSignalsConfig.h'])
 
 pkg_mod = import('pkgconfig')
 pkg_mod.generate(libraries: lib,
                  version: '1.0',
                  name: 'libqtappfw-vehicle-signals',
                  filebase: 'qtappfw-vehicle-signals',
-                 requires: ['Qt5Core', 'Qt5WebSockets'],
-                 description: 'Library wrapping VIS vehicle signaling API in Qt objects')
+                 requires: ['Qt5Core'],
+                 description: 'Library wrapping VSS API in Qt objects')
 
 qtappfw_vs_dep = declare_dependency(dependencies: qt5_dep,
                                     link_with: lib,
diff --git a/vehicle-signals/vehiclesignals.cpp b/vehicle-signals/vehiclesignals.cpp
index d6cf8eb..adf371a 100644
--- a/vehicle-signals/vehiclesignals.cpp
+++ b/vehicle-signals/vehiclesignals.cpp
@@ -1,427 +1,176 @@
 /*
- * Copyright (C) 2022 Konsulko Group
+ * Copyright (C) 2022,2023 Konsulko Group
  *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * SPDX-License-Identifier: Apache-2.0
  */
 
 #include <QDebug>
-#include <QSettings>
-#include <QUrl>
-#include <QFile>
-#include <QSslKey>
-#include <QTimer>
-#include <QVariantMap>
-#include <QJsonDocument>
-#include <QJsonObject>
+#include <QtConcurrent>
 
 #include "vehiclesignals.h"
+#include "QtKuksaClient.h"
 
-#define DEFAULT_CLIENT_KEY_FILE  "/etc/kuksa-val/Client.key"
-#define DEFAULT_CLIENT_CERT_FILE "/etc/kuksa-val/Client.pem"
-#define DEFAULT_CA_CERT_FILE     "/etc/kuksa-val/CA.pem"
-
-VehicleSignalsConfig::VehicleSignalsConfig(const QString &hostname,
-					   const unsigned port,
-					   const QByteArray &clientKey,
-					   const QByteArray &clientCert,
-					   const QByteArray &caCert,
-					   const QString &authToken,
-					   bool verifyPeer) :
-	m_hostname(hostname),
-	m_port(port),
-	m_clientKey(clientKey),
-	m_clientCert(clientCert),
-	m_caCert(caCert),
-	m_authToken(authToken),
-	m_verifyPeer(verifyPeer),
-	m_verbose(0),
-	m_valid(true)
-{
-	// Potentially could do some certificate validation here...
-}
-
-VehicleSignalsConfig::VehicleSignalsConfig(const QString &appname)
+VehicleSignals::VehicleSignals(const VehicleSignalsConfig &config, QObject *parent) :
+	QObject(parent),
+	m_config(config)
 {
-	m_valid = false;
-
-	QSettings *pSettings = new QSettings("AGL", appname);
-	if (!pSettings)
-		return;
-
-	m_hostname = pSettings->value("vis-client/server", "localhost").toString();
-	if (m_hostname.isEmpty()) {
-		qCritical() << "Invalid server hostname";
-		return;
-	}
-
-	m_port = pSettings->value("vis-client/port", 8090).toInt();
-	if (m_port == 0) {
-		qCritical() << "Invalid server port";
-		return;
-	}
-
-	// Default to disabling peer verification for now to be able
-        // to use the default upstream KUKSA.val certificates for
-	// testing.  Wrangling server and CA certificate generation
-	// and management to be able to verify will require further
-	// investigation.
-	m_verifyPeer = pSettings->value("vis-client/verify-server", false).toBool();
-
-	QString keyFileName = pSettings->value("vis-client/key", DEFAULT_CLIENT_KEY_FILE).toString();
-	if (keyFileName.isEmpty()) {
-		qCritical() << "Invalid client key filename";
-		return;
-	}
-	QFile keyFile(keyFileName);
-	if (!keyFile.open(QIODevice::ReadOnly)) {
-		qCritical() << "Could not open client key file";
-		return;
-	}
-	QByteArray keyData = keyFile.readAll();
-	if (keyData.isEmpty()) {
-		qCritical() << "Invalid client key file";
-		return;
-	}
-	m_clientKey = keyData;
-
-	QString certFileName = pSettings->value("vis-client/certificate", DEFAULT_CLIENT_CERT_FILE).toString();
-	if (certFileName.isEmpty()) {
-		qCritical() << "Invalid client certificate filename";
-		return;
-	}
-	QFile certFile(certFileName);
-	if (!certFile.open(QIODevice::ReadOnly)) {
-		qCritical() << "Could not open client certificate file";
-		return;
-	}
-	QByteArray certData = certFile.readAll();
-	if (certData.isEmpty()) {
-		qCritical() << "Invalid client certificate file";
-		return;
-	}
-	m_clientCert = certData;
-
-	QString caCertFileName = pSettings->value("vis-client/ca-certificate", DEFAULT_CA_CERT_FILE).toString();
-	if (caCertFileName.isEmpty()) {
-		qCritical() << "Invalid CA certificate filename";
-		return;
-	}
-	QFile caCertFile(caCertFileName);
-	if (!caCertFile.open(QIODevice::ReadOnly)) {
-		qCritical() << "Could not open CA certificate file";
-		return;
-	}
-	QByteArray caCertData = caCertFile.readAll();
-	if (caCertData.isEmpty()) {
-		qCritical() << "Invalid CA certificate file";
-		return;
-	}
-	// Pre-check CA certificate
-	QList<QSslCertificate> newSslCaCerts = QSslCertificate::fromData(caCertData);
-	if (newSslCaCerts.isEmpty()) {
-		qCritical() << "Invalid CA certificate";
-		return;
-	}
-	m_caCert = caCertData;
-
-	QString authTokenFileName = pSettings->value("vis-client/authorization").toString();
-	if (authTokenFileName.isEmpty()) {
-		qCritical() << "Invalid authorization token filename";
-		return;
-	}
-	QFile authTokenFile(authTokenFileName);
-	if (!authTokenFile.open(QIODevice::ReadOnly | QIODevice::Text)) {
-		qCritical() << "Could not open authorization token file";
-		return;
-	}
-	QTextStream in(&authTokenFile);
-	QString authToken = in.readLine();
-	if (authToken.isEmpty()) {
-		qCritical() << "Invalid authorization token file";
-		return;
+	// Create gRPC channel
+	// NOTE: channel creation and waiting for connected state could be put into
+	//       a thread that is spawned here.
+
+  	QString host = m_config.hostname();
+	host += ":";
+	host += QString::number(m_config.port());
+
+	std::shared_ptr<grpc::Channel> channel;
+	if (!m_config.caCert().isEmpty()) {
+		qInfo() << "Using TLS";
+		grpc::SslCredentialsOptions options;
+		options.pem_root_certs = m_config.caCert().toStdString();
+		if (!m_config.tlsServerName().isEmpty()) {
+			grpc::ChannelArguments args;
+			auto target = m_config.tlsServerName();
+			qInfo() << "Overriding TLS target name with " << target;
+			args.SetString(GRPC_SSL_TARGET_NAME_OVERRIDE_ARG, target.toStdString());
+			channel = grpc::CreateCustomChannel(host.toStdString(), grpc::SslCredentials(options), args);
+		} else {
+			channel = grpc::CreateChannel(host.toStdString(), grpc::SslCredentials(options));
+		}
+	} else {
+		channel = grpc::CreateChannel(host.toStdString(), grpc::InsecureChannelCredentials());
 	}
-	m_authToken = authToken;
 
-	m_verbose = 0;
-	QString verbose = pSettings->value("vis-client/verbose").toString();
-	if (!verbose.isEmpty()) {
-		if (verbose == "true" || verbose == "1")
-			m_verbose = 1;
-		if (verbose == "2")
-			m_verbose = 2;
-	}
+	m_broker = new QtKuksaClient(channel, config);
+	if (!m_broker)
+		qCritical() << "gRPC client initialization failed";
 
-	m_valid = true;
-}
-
-VehicleSignals::VehicleSignals(const VehicleSignalsConfig &config, QObject *parent) :
-	QObject(parent),
-	m_config(config),
-	m_request_id(0)
-{
-	QObject::connect(&m_websocket, &QWebSocket::connected, this, &VehicleSignals::onConnected);
-	QObject::connect(&m_websocket, QOverload<QAbstractSocket::SocketError>::of(&QWebSocket::error),
-			 this, &VehicleSignals::onError);
-	QObject::connect(&m_websocket, &QWebSocket::disconnected, this, &VehicleSignals::onDisconnected);
+	QObject::connect(m_broker, &QtKuksaClient::connected, this, &VehicleSignals::onConnected);
 }
 
 VehicleSignals::~VehicleSignals()
 {
-	m_websocket.close();
+	delete m_broker;
 }
 
 void VehicleSignals::connect()
 {
-	if (!m_config.valid()) {
-		qCritical() << "Invalid VIS server configuration";
-		return;
-	}
-
-	QUrl visUrl;
-	visUrl.setScheme(QStringLiteral("wss"));
-	visUrl.setHost(m_config.hostname());
-	visUrl.setPort(m_config.port());
-
-	QSslConfiguration sslConfig = QSslConfiguration::defaultConfiguration();
-
-	// Add client private key
-        // i.e. kuksa_certificates/Client.key in source tree
-	QSslKey sslKey(m_config.clientKey(), QSsl::Rsa);
-	sslConfig.setPrivateKey(sslKey);
-
-	// Add local client certificate
-        // i.e. kuksa_certificates/Client.pem in source tree
-	QList<QSslCertificate> sslCerts = QSslCertificate::fromData(m_config.clientCert());
-	if (sslCerts.empty()) {
-		qCritical() << "Invalid client certificate";
-		return; 
-	}
-	sslConfig.setLocalCertificate(sslCerts.first());
-
-	// Add CA certificate
-        // i.e. kuksa_certificates/CA.pem in source tree
-	// Note the following can be simplified with QSslConfiguration::addCaCertificate with Qt 5.15
-	QList<QSslCertificate> sslCaCerts = sslConfig.caCertificates();
-	QList<QSslCertificate> newSslCaCerts = QSslCertificate::fromData(m_config.caCert());
-	if (newSslCaCerts.empty()) {
-		qCritical() << "Invalid CA certificate";
-		return;
-	}
-	sslCaCerts.append(newSslCaCerts.first());
-	sslConfig.setCaCertificates(sslCaCerts);
-
-	sslConfig.setPeerVerifyMode(m_config.verifyPeer() ? QSslSocket::VerifyPeer : QSslSocket::VerifyNone);
+	// QtKuksaClient will call our onConnected slot when the channel
+	// is connected, and then we pass that along via our connected
+	// signal.
+	if (m_broker)
+		m_broker->connect();
+}
 
-	m_websocket.setSslConfiguration(sslConfig);
+void VehicleSignals::authorize()
+{
+	// Databroker has no separate authorize call, so this is a no-op
+	emit authorized();
+}
 
-	if (m_config.verbose())
-		qInfo() << "Opening VIS websocket";
-	m_websocket.open(visUrl);
+void VehicleSignals::get(const QString &path, const bool actuator)
+{
+	if (m_broker)
+		m_broker->get(path, actuator);
 }
 
-void VehicleSignals::onConnected()
+void VehicleSignals::set(const QString &path, const QString &value, const bool actuator)
 {
-	if (m_config.verbose() > 1)
-		qDebug() << "VehicleSignals::onConnected: enter";
-	QObject::connect(&m_websocket, &QWebSocket::textMessageReceived, this, &VehicleSignals::onTextMessageReceived);
-	emit connected();
+	if (m_broker)
+		m_broker->set(path, value, actuator);
 }
 
-void VehicleSignals::onError(QAbstractSocket::SocketError error)
+void VehicleSignals::set(const QString &path, const qint32 value, const bool actuator)
 {
-	if (m_config.verbose() > 1)
-		qDebug() << "VehicleSignals::onError: enter";
-	QTimer::singleShot(1000, this, &VehicleSignals::reconnect);
+	if (m_broker)
+		m_broker->set(path, value, actuator);
 }
 
-void VehicleSignals::reconnect()
+void VehicleSignals::set(const QString &path, const qint64 value, const bool actuator)
 {
-	if (m_config.verbose() > 1)
-		qDebug() << "VehicleSignals::reconnect: enter";
-	connect();
+	if (m_broker)
+		m_broker->set(path, value, actuator);
 }
 
-void VehicleSignals::onDisconnected()
+void VehicleSignals::set(const QString &path, const quint32 value, const bool actuator)
 {
-	if (m_config.verbose() > 1)
-		qDebug() << "VehicleSignals::onDisconnected: enter";
-	QObject::disconnect(&m_websocket, &QWebSocket::textMessageReceived, this, &VehicleSignals::onTextMessageReceived);
-	emit disconnected();
+	if (m_broker)
+		m_broker->set(path, value, actuator);
+}
 
-	// Try to reconnect
-	QTimer::singleShot(1000, this, &VehicleSignals::reconnect);
+void VehicleSignals::set(const QString &path, const quint64 value, const bool actuator)
+{
+	if (m_broker)
+		m_broker->set(path, value, actuator);
 }
 
-void VehicleSignals::authorize()
+void VehicleSignals::set(const QString &path, const float value, const bool actuator)
 {
-	QVariantMap map;
-	map["action"] = QString("authorize");
-	map["tokens"] = m_config.authToken();
-	map["requestId"] = QString::number(m_request_id++);
-	QJsonDocument doc = QJsonDocument::fromVariant(map);
-	m_websocket.sendTextMessage(doc.toJson(QJsonDocument::Compact).data());
+	if (m_broker)
+		m_broker->set(path, value, actuator);
 }
 
-void VehicleSignals::get(const QString &path)
+void VehicleSignals::set(const QString &path, const double value, const bool actuator)
 {
-	QVariantMap map;
-	map["action"] = QString("get");
-	map["tokens"] = m_config.authToken();
-	map["path"] = path;
-	map["requestId"] = QString::number(m_request_id++);
-	QJsonDocument doc = QJsonDocument::fromVariant(map);
-	m_websocket.sendTextMessage(doc.toJson(QJsonDocument::Compact).data());
+	if (m_broker)
+		m_broker->set(path, value, actuator);
 }
 
-void VehicleSignals::set(const QString &path, const QString &value)
+void VehicleSignals::subscribe(const QString &path, bool actuator)
 {
-	QVariantMap map;
-	map["action"] = QString("set");
-	map["tokens"] = m_config.authToken();
-	map["path"] = path;
-	map["value"] = value;
-	map["requestId"] = QString::number(m_request_id++);
-	QJsonDocument doc = QJsonDocument::fromVariant(map);
-	m_websocket.sendTextMessage(doc.toJson(QJsonDocument::Compact).data());
+	if (m_broker)
+		m_broker->subscribe(path, actuator);
 }
 
-void VehicleSignals::subscribe(const QString &path)
+void VehicleSignals::subscribe(const QMap<QString, bool> &signals_)
 {
-	QVariantMap map;
-	map["action"] = QString("subscribe");
-	map["tokens"] = m_config.authToken();
-	map["path"] = path;
-	map["requestId"] = QString::number(m_request_id++);
-	QJsonDocument doc = QJsonDocument::fromVariant(map);
-	m_websocket.sendTextMessage(doc.toJson(QJsonDocument::Compact).data());
+	if (m_broker)
+		m_broker->subscribe(signals_);
 }
 
-bool VehicleSignals::parseData(const QJsonObject &response, QString &path, QString &value, QString &timestamp)
+// Slots
+
+void VehicleSignals::onConnected()
 {
-	if (response.contains("error")) {
-		QString error = response.value("error").toString();
-		return false;
-	}
+	QObject::connect(m_broker, &QtKuksaClient::getResponse, this, &VehicleSignals::onGetResponse);
+	QObject::connect(m_broker, &QtKuksaClient::setResponse, this, &VehicleSignals::onSetResponse);
+	QObject::connect(m_broker, &QtKuksaClient::subscribeResponse, this, &VehicleSignals::onSubscribeResponse);
+	//QObject::connect(m_broker, &QtKuksaClient::subscribeDone, this, &VehicleSignals::onSubscribeDone);
 
-	if (!(response.contains("data") && response["data"].isObject())) {
-		qWarning() << "Malformed response (data missing)";
-		return false;
-	}
-	QJsonObject data = response["data"].toObject();
-	if (!(data.contains("path") && data["path"].isString())) {
-		qWarning() << "Malformed response (path missing)";
-		return false;
-	}
-	path = data["path"].toString();
-	// Convert '/' to '.' in paths to ensure consistency for clients
-	path.replace(QString("/"), QString("."));
+	emit connected();
+}
 
-	if (!(data.contains("dp") && data["dp"].isObject())) {
-		qWarning() << "Malformed response (datapoint missing)";
-		return false;
-	}
-	QJsonObject dp = data["dp"].toObject();
-	if (!dp.contains("value")) {
-		qWarning() << "Malformed response (value missing)";
-		return false;
-	} else if (dp["value"].isString()) {
-		value = dp["value"].toString();
-	} else if (dp["value"].isDouble()) {
-		value = QString::number(dp["value"].toDouble(), 'f', 9);
-	} else if (dp["value"].isBool()) {
-		value = dp["value"].toBool() ? "true" : "false";
-	} else {
-		qWarning() << "Malformed response (unsupported value type)";
-		return false;
-	}
+void VehicleSignals::onGetResponse(QString path, QString value, QString timestamp)
+{
+	emit getSuccessResponse(path, value, timestamp);
+}
 
-	if (!(dp.contains("ts") && dp["ts"].isString())) {
-		qWarning() << "Malformed response (timestamp missing)";
-		return false;
-	}
-	timestamp = dp["ts"].toString();
+void VehicleSignals::onSetResponse(QString path, QString error)
+{
+	emit setErrorResponse(path, error);
+}
 
-	return true;
+void VehicleSignals::onSubscribeResponse(QString path, QString value, QString timestamp)
+{
+	if (m_config.verbose() > 1)
+		qDebug() << "VehicleSignals::onSubscribeResponse: got " << path << " = " << value;
+	emit signalNotification(path, value, timestamp);
 }
 
-//
-// NOTE:
-//
-// Ideally request ids would be used to provide some form of mapping
-// to callers of get/set for responses/errors.  At present the demo
-// usecases are simple enough that it does not seem worth implementing
-// just yet.
-//
-void VehicleSignals::onTextMessageReceived(QString msg)
+void VehicleSignals::onSubscribeDone(const QMap<QString, bool> &signals_, bool canceled)
 {
-	msg = msg.simplified();
-	QJsonDocument doc(QJsonDocument::fromJson(msg.toUtf8()));
-	if (doc.isEmpty()) {
-		qWarning() << "Received invalid JSON: empty VIS message";
-		return;
+	if (!canceled) {
+		// queue up a resubscribe attempt
+		QFuture<void> future = QtConcurrent::run(this, &VehicleSignals::resubscribe, signals_);
 	}
+}
 
-	if (!doc.isObject()) {
-		qWarning() << "Received invalid JSON: malformed VIS message";
-		return;
-	}
-	QJsonObject obj = doc.object();
+// Private
 
-	if (!obj.contains("action")) {
-		qWarning() << "Received unknown message (no action), discarding";
-		return;
-	}
-	
-	QString action = obj.value("action").toString();
-	if (action == "authorize") {
-		if (obj.contains("error")) {
-			QString error = obj.value("error").toString();
-			qWarning() << "VIS authorization failed: " << error;
-		} else {
-			if (m_config.verbose() > 1)
-				qDebug() << "authorized";
-			emit authorized();
-		}
-	} else if (action == "subscribe") {
-		if (obj.contains("error")) {
-			QString error = obj.value("error").toString();
-			qWarning() << "VIS subscription failed: " << error;
-		}
-	} else if (action == "get") {
-		if (obj.contains("error")) {
-			QString error = obj.value("error").toString();
-			qWarning() << "VIS get failed: " << error;
-		} else {
-			QString path, value, ts;
-			if (parseData(obj, path, value, ts)) {
-				if (m_config.verbose() > 1)
-					qDebug() << "VehicleSignals::onTextMessageReceived: emitting response" << path << " = " << value;
-				emit getSuccessResponse(path, value, ts);
-			}
-		}
-	} else if (action == "set") {
-		if (obj.contains("error")) {
-			QString error = obj.value("error").toString();
-			qWarning() << "VIS set failed: " << error;
-		}
-	} else if (action == "subscription") {
-		QString path, value, ts;
-		if (parseData(obj, path, value, ts)) {
-			if (m_config.verbose() > 1)
-				qDebug() << "VehicleSignals::onTextMessageReceived: emitting notification" << path << " = " << value;
-			emit signalNotification(path, value, ts);
-		}
-	} else {
-		qWarning() << "unhandled VIS response of type: " << action;
-	}
+void VehicleSignals::resubscribe(const QMap<QString, bool> &signals_)
+{
+	// Delay 100 milliseconds between subscribe attempts
+	QThread::msleep(100);
+
+	if (m_broker)
+		m_broker->subscribe(signals_);
 }
diff --git a/vehicle-signals/vehiclesignals.h b/vehicle-signals/vehiclesignals.h
index fd5fe9c..734cb27 100644
--- a/vehicle-signals/vehiclesignals.h
+++ b/vehicle-signals/vehiclesignals.h
@@ -1,63 +1,19 @@
 /*
- * Copyright (C) 2022 Konsulko Group
+ * Copyright (C) 2022,2023 Konsulko Group
  *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
+ * SPDX-License-Identifier: Apache-2.0
  */
 
 #ifndef VEHICLESIGNALS_H
 #define VEHICLESIGNALS_H
 
 #include <QObject>
-#include <QWebSocket>
 
-// Class to read/hold VIS server configuration
+#include "VehicleSignalsConfig.h"
 
-class VehicleSignalsConfig
-{
-public:
-        explicit VehicleSignalsConfig(const QString &hostname,
-				      const unsigned port,
-				      const QByteArray &clientKey,
-				      const QByteArray &clientCert,
-				      const QByteArray &caCert,
-				      const QString &authToken,
-				      bool verifyPeer = true);
-        explicit VehicleSignalsConfig(const QString &appname);
-        ~VehicleSignalsConfig() {};
-
-	QString hostname() { return m_hostname; };
-	unsigned port() { return m_port; };
-	QByteArray clientKey() { return m_clientKey; };
-	QByteArray clientCert() { return m_clientCert; };
-	QByteArray caCert() { return m_caCert; };
-	QString authToken() { return m_authToken; };
-	bool verifyPeer() { return m_verifyPeer; };
-	bool valid() { return m_valid; };
-	unsigned verbose() { return m_verbose; };
+class QtKuksaClient;
 
-private:
-	QString m_hostname;
-	unsigned m_port;
-	QByteArray m_clientKey;
-	QByteArray m_clientCert;
-	QByteArray m_caCert;
-	QString m_authToken;
-	bool m_verifyPeer;
-	bool m_valid;
-	unsigned m_verbose;
-};
-
-// VIS signaling interface class
+// VSS signal interface class
 
 class VehicleSignals : public QObject
 {
@@ -70,30 +26,39 @@ public:
 	Q_INVOKABLE void connect();	
 	Q_INVOKABLE void authorize();
 
-	Q_INVOKABLE void get(const QString &path);
-	Q_INVOKABLE void set(const QString &path, const QString &value);
-	Q_INVOKABLE void subscribe(const QString &path);
+	Q_INVOKABLE void get(const QString &path, const bool actuator = false);
+
+	Q_INVOKABLE void set(const QString &path, const QString &value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const qint32 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const qint64 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const quint32 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const quint64 value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const float value, const bool actuator = false);
+	Q_INVOKABLE void set(const QString &path, const double value, const bool actuator = false);
+
+	Q_INVOKABLE void subscribe(const QString &path, const bool actuator = false);
+	Q_INVOKABLE void subscribe(const QMap<QString, bool> &signals_);
 
 signals:
 	void connected();
 	void authorized();
-        void getSuccessResponse(QString path, QString value, QString timestamp);
-        void signalNotification(QString path, QString value, QString timestamp);
 	void disconnected();
+	void getSuccessResponse(QString path, QString value, QString timestamp);
+	void setErrorResponse(QString path, QString error);
+	void signalNotification(QString path, QString value, QString timestamp);
 
 private slots:
 	void onConnected();
-	void onError(QAbstractSocket::SocketError error);
-	void reconnect();
-	void onDisconnected();
-	void onTextMessageReceived(QString message);
+	void onGetResponse(QString path, QString value, QString timestamp);
+	void onSetResponse(QString path, QString error);
+	void onSubscribeResponse(QString path, QString value, QString timestamp);
+	void onSubscribeDone(const QMap<QString, bool> &signals_, bool canceled);
 
 private:
 	VehicleSignalsConfig m_config;
-	QWebSocket m_websocket;
-	std::atomic<unsigned int> m_request_id;
+	QtKuksaClient *m_broker;
 
-	bool parseData(const QJsonObject &response, QString &path, QString &value, QString &timestamp);
+	void resubscribe(const QMap<QString, bool> &signals_);
 };
 
 #endif // VEHICLESIGNALS_H