Initial code check-inicefish_8.99.5icefish_8.99.4icefish_8.99.3icefish_8.99.2icefish_8.99.1icefish/8.99.5icefish/8.99.4icefish/8.99.3icefish/8.99.2icefish/8.99.1halibut_8.0.6halibut_8.0.5halibut_8.0.4halibut_8.0.3halibut_8.0.2halibut_8.0.1halibut_8.0.0halibut_7.99.3halibut_7.99.2halibut_7.99.1halibut/8.0.6halibut/8.0.5halibut/8.0.4halibut/8.0.3halibut/8.0.2halibut/8.0.1halibut/8.0.0halibut/7.99.3halibut/7.99.2halibut/7.99.18.99.58.99.48.99.38.99.28.99.18.0.68.0.58.0.48.0.38.0.28.0.18.0.07.99.37.99.27.99.1halibut

A simple telematics demo application for AGL.  It reads vehicle and
engine speed from the CAN low-level binding and publishes them via
MQTT.

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

1 files changed, 236 insertions, 0 deletions

diff --git a/app/afbclient.cpp b/app/afbclient.cpp
new file mode 100644
index 0000000..a40f6c9
--- /dev/null
+++ b/app/afbclient.cpp
@@ -0,0 +1,236 @@
+/*
+ * Copyright (C) 2019 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.
+ */
+
+#include "afbclient.h"
+#include <string>
+#include <cstring>
+#include <iostream>
+#include <mutex>
+#include <condition_variable>
+
+#undef DEBUG
+//#define DEBUG
+
+struct call_data
+{
+	bool sync;
+	std::mutex mutex;
+	std::condition_variable cv;
+	bool ready;
+	std::function<void(json_object*)> cb;
+	json_object *resp;
+};
+
+static void on_hangup_cb(void *closure, struct afb_wsj1 *wsj)
+{
+}
+
+static void on_call_cb(void *closure, const char *api, const char *verb, struct afb_wsj1_msg *msg)
+{
+}
+
+static void on_reply_cb(void *closure, struct afb_wsj1_msg *msg)
+{
+	call_data *data = (call_data*) closure;
+	struct json_object* reply;
+
+	if(!data)
+		goto reply_done;
+
+	reply = afb_wsj1_msg_object_j(msg);
+	if(reply) {
+#ifdef DEBUG
+		std::cerr << __FUNCTION__ << ": reply = " << \
+			json_object_to_json_string_ext(reply, JSON_C_TO_STRING_SPACED | JSON_C_TO_STRING_PRETTY) << \
+			std::endl;
+#endif
+		if(data->sync) {
+			data->resp = reply;
+
+			// Increase reference count since we are going to use
+			// reply after this callback returns, caller must do a
+			// put.
+			json_object_get(reply);
+		} else if(data->cb != nullptr) {
+			data->cb(reply);
+		}
+	}
+reply_done:
+	if(data->sync) {
+		// Signal reply is done
+		{
+			std::lock_guard<std::mutex> lk(data->mutex);
+			data->ready = true;
+		}
+		data->cv.notify_one();
+	}
+}
+
+//
+// on_event_cb is inline in afbclient.h
+//
+
+static void *afb_loop_thread(struct sd_event* loop)
+{
+	for(;;)
+		sd_event_run(loop, 30000000);
+}
+
+AfbClient::AfbClient(const int port, const std::string &token)
+{
+	std::string uri;
+
+	if(sd_event_new(&m_afb_loop) < 0) {
+		std::cerr << __FUNCTION__ << ": Failed to create event loop" << std::endl;
+		return;
+	}
+
+	// Initialize interface for websocket
+	m_itf.on_hangup = on_hangup_cb;
+	m_itf.on_call = on_call_cb;
+	m_itf.on_event = on_event_cb;
+
+	uri = "ws://localhost:" + std::to_string(port) + "/api?token=" + token;
+#ifdef DEBUG
+	std::cerr << "Using URI: " << uri << std::endl;
+#endif
+	m_ws = afb_ws_client_connect_wsj1(m_afb_loop, uri.c_str(), &m_itf, this);
+	if(m_ws) {
+		m_afb_thread = std::thread(afb_loop_thread, m_afb_loop);
+	} else {
+		std::cerr << __FUNCTION__ << ": Failed to create websocket connection" << std::endl;
+		goto error;
+	}
+
+	m_valid = true;
+	return;
+error:
+	if(m_afb_loop) {
+		sd_event_unref(m_afb_loop);
+		m_afb_loop = nullptr;
+	}
+	return;
+}
+
+AfbClient::~AfbClient(void)
+{
+	sd_event_unref(m_afb_loop);
+	afb_wsj1_unref(m_ws);
+}
+
+int AfbClient::call(const std::string &api, const std::string &verb, struct json_object *arg, callback_fn cb)
+{
+	if(!m_valid)
+		return -1;
+
+	call_data data;
+	data.sync = false;
+	data.cb = cb;
+	int rc = afb_wsj1_call_j(m_ws, api.c_str(), verb.c_str(), arg, on_reply_cb, (void*) &data);
+	if(rc < 0) {
+		std::cerr << __FUNCTION__ << \
+			": Failed to call " << \
+			api.c_str() << \
+			"/" << \
+			verb.c_str() << \
+			std::endl;
+	}
+	return rc;
+}
+
+int AfbClient::call_sync(const std::string &api, const std::string &verb, struct json_object *arg, struct json_object **resp)
+{
+	if(!m_valid)
+		return -1;
+
+	call_data data;
+	data.sync = true;
+	data.ready = false;
+	data.cb = nullptr;
+	int rc = afb_wsj1_call_j(m_ws, api.c_str(), verb.c_str(), arg, on_reply_cb, (void*) &data);
+	if(rc >= 0) {
+		// Wait for response
+		std::unique_lock<std::mutex> lk(data.mutex);
+		data.cv.wait(lk, [&]{ return data.ready; });
+
+		if(resp && data.resp)
+			*resp = data.resp;
+	} else {
+		std::cerr << __FUNCTION__ << \
+			": Failed to call " << \
+			api.c_str() << \
+			"/" << \
+			verb.c_str() << \
+			std::endl;
+	}
+	return rc;
+}
+
+int AfbClient::subscribe(const std::string &api, const std::string &event, const std::string &eventString, callback_fn cb, const std::string &eventValueString)
+{
+	if(!m_valid)
+		return -1;
+
+	// For now, simply let the user over-write the callback if they
+	// specify the same eventString.  Avoiding that and keeping track
+	// of the duplicate eventStrings/callbacks will complicate things
+	// quite a bit.
+
+        struct json_object *j_obj = json_object_new_object();
+        json_object_object_add(j_obj, eventValueString.c_str(), json_object_new_string(event.c_str()));
+        int rc = call_sync(api, std::string("subscribe"), j_obj);
+	if(rc >= 0 && cb != nullptr) {
+		m_event_handlers[eventString] = cb;
+	}
+	return rc;
+}
+
+int AfbClient::unsubscribe(const std::string &api, const std::string &eventString, const std::string &eventValueString)
+{
+	if(!m_valid)
+		return -1;
+
+	if(m_event_handlers.find(eventString) == m_event_handlers.end())
+		return -1;
+
+        struct json_object *j_obj = json_object_new_object();
+        json_object_object_add(j_obj, eventValueString.c_str(), json_object_new_string(eventString.c_str()));
+        int rc = call_sync(api, std::string("unsubscribe"), j_obj);
+	if(rc >= 0) {
+		m_event_handlers.erase(eventString);
+	}
+	return rc;
+}
+
+void AfbClient::on_event(const char* event, struct afb_wsj1_msg *msg)
+{
+#if 0
+	std::cerr << __FUNCTION__ << ": event = " << event << std::endl;
+#endif
+	struct json_object *contents = afb_wsj1_msg_object_j(msg);
+#ifdef DEBUG
+	std::cerr << __FUNCTION__ << ": contents = " << \
+		json_object_to_json_string_ext(contents, JSON_C_TO_STRING_SPACED | JSON_C_TO_STRING_PRETTY) << \
+		std::endl;
+#endif
+	struct json_object *data;
+	if(json_object_object_get_ex(contents, "data", &data)) {
+		auto i = m_event_handlers.find(std::string(event));
+		if (i != m_event_handlers.end() && i->second != nullptr) {
+			i->second(data);
+		}
+	}
+}