Adding also diagnostic request to systemd event loop

As for CAN signal, monitoring diagnostic request messages is now handled by
systemD io event loop. Socket reading is common for all OBD2 signals and handled
by the diagnostic manager.
systemd callback function lies in binding callback which in turns call read_socket
method of diagnostic manager. Processing is little bit different from classic CAN
messages so it is a separate callback function.

Lot of cleaning to do now...

Change-Id: I4d2ada0beb5d3348736dfdf3c56a7cb875a1c6c7
Signed-off-by: Romain Forlot <romain.forlot@iot.bzh>

4 files changed, 48 insertions, 11 deletions

diff --git a/CAN-binder/low-can-binding/binding/low-can-cb.cpp b/CAN-binder/low-can-binding/binding/low-can-cb.cpp
index 55d9bdd4..b3c9e2e0 100644
--- a/CAN-binder/low-can-binding/binding/low-can-cb.cpp
+++ b/CAN-binder/low-can-binding/binding/low-can-cb.cpp
@@ -52,7 +52,7 @@ void on_no_clients(std::string message)
 	}
 }
 
-int read(sd_event_source *s, int fd, uint32_t revents, void *userdata)
+int read_can_signal(sd_event_source *s, int fd, uint32_t revents, void *userdata)
 {
 	can_signal_t* sig= (can_signal_t*)userdata;
 	sig->read_socket();
@@ -67,6 +67,25 @@ int read(sd_event_source *s, int fd, uint32_t revents, void *userdata)
 	return 0;
 }
 
+int read_diagnostic_message(sd_event_source *s, int fd, uint32_t revents, void *userdata)
+{
+	diagnostic_manager_t& diag_m = configuration_t::instance().get_diagnostic_manager();
+	diag_m.read_socket();
+
+	/* check if error or hangup */
+	if ((revents & (EPOLLERR|EPOLLRDHUP|EPOLLHUP)) != 0)
+	{
+		sd_event_source_unref(s);
+		diag_m.get_socket().close();
+		diag_m.cleanup_active_requests(true);
+		ERROR(binder_interface, "%s: Error on diagnostic manager socket, cancelling active requests.", __FUNCTION__);
+		return -1;
+	}
+
+	return 0;
+}
+
+
 ///******************************************************************************
 ///
 ///		Subscription and unsubscription
@@ -161,10 +180,9 @@ static int subscribe_unsubscribe_signals(struct afb_req request, bool subscribe,
 		// poll a PID for nothing.
 		if(sig->get_supported() && subscribe)
 		{
-				float frequency = sig->get_frequency();
-				subscribe = diag_m.add_recurring_request(
-					diag_req, sig->get_name().c_str(), false, sig->get_decoder(), sig->get_callback(), (float)frequency);
-					//TODO: Adding callback requesting ignition status:	diag_req, sig.c_str(), false, diagnostic_message_t::decode_obd2_response, diagnostic_message_t::check_ignition_status, frequency);
+			float frequency = sig->get_frequency();
+			diag_m.add_recurring_request(diag_req, sig->get_name().c_str(), false, sig->get_decoder(), sig->get_callback(), (float)frequency);
+				//TODO: Adding callback requesting ignition status:	diag_req, sig.c_str(), false, diagnostic_message_t::decode_obd2_response, diagnostic_message_t::check_ignition_status, frequency);
 		}
 		else
 		{
@@ -188,8 +206,7 @@ static int subscribe_unsubscribe_signals(struct afb_req request, bool subscribe,
 		{
 			return -1;
 		}
-		struct sd_event_source* e_source;
-		sd_event_add_io(afb_daemon_get_event_loop(binder_interface->daemon), &e_source, sig->get_socket().socket(), EPOLLIN, read, sig.get());
+		sd_event_add_io(afb_daemon_get_event_loop(binder_interface->daemon), &e_source, sig->get_socket().socket(), EPOLLIN, read_can_signal, sig.get());
 		rets++;
 		DEBUG(binder_interface, "%s: signal: %s subscribed", __FUNCTION__, sig->get_name().c_str());
 	}
diff --git a/CAN-binder/low-can-binding/binding/low-can-hat.hpp b/CAN-binder/low-can-binding/binding/low-can-hat.hpp
index 7df4663c..57c0a536 100644
--- a/CAN-binder/low-can-binding/binding/low-can-hat.hpp
+++ b/CAN-binder/low-can-binding/binding/low-can-hat.hpp
@@ -20,6 +20,7 @@
 
 #include <cstddef>
 #include <string>
+#include <systemd/sd-event.h>
 
 extern "C"
 {
@@ -31,6 +32,8 @@ extern "C" struct afb_binding_interface;
 extern const struct afb_binding_interface *binder_interface;
 
 void on_no_clients(std::string message);
+int read_can_signal(sd_event_source *s, int fd, uint32_t revents, void *userdata);
+int read_diagnostic_message(sd_event_source *s, int fd, uint32_t revents, void *userdata);
 
 void subscribe(struct afb_req request);
 void unsubscribe(struct afb_req request);
\ No newline at end of file
diff --git a/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.cpp b/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.cpp
index 8a55365e..935767a8 100644
--- a/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.cpp
+++ b/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.cpp
@@ -49,6 +49,7 @@ bool diagnostic_manager_t::initialize()
 	bus_ = configuration_t::instance().get_diagnostic_bus();
 
 	init_diagnostic_shims();
+	event_source_ = nullptr;
 	reset();
 
 	initialized_ = true;
@@ -456,15 +457,28 @@ active_diagnostic_request_t* diagnostic_manager_t::add_recurring_request(Diagnos
 		{
 			// TODO: implement Acceptance Filter
 			//if(updateRequiredAcceptanceFilters(bus, request)) {
-			active_diagnostic_request_t* entry = new active_diagnostic_request_t(bus_, request, name,
+			entry = new active_diagnostic_request_t(bus_, request, name,
 					wait_for_multiple_responses, decoder, callback, frequencyHz);
 			recurring_requests_.push_back(entry);
 
 			entry->set_handle(shims_, request);
 			if(add_rx_filter(OBD2_FUNCTIONAL_BROADCAST_ID) < 0)
-			{ recurring_requests_.pop_back(); }
+				{ recurring_requests_.pop_back(); }
 			else
-			{ start_diagnostic_request(&shims_, entry->get_handle()); }
+				{
+					start_diagnostic_request(&shims_, entry->get_handle()); 
+					if(event_source_ == nullptr && sd_event_add_io(afb_daemon_get_event_loop(binder_interface->daemon), 
+						&event_source_,
+						socket_.socket(),
+						EPOLLIN,
+						read_diagnostic_message,
+						nullptr) < 0)
+					{
+						cleanup_request(entry, true);
+						WARNING(binder_interface, "%s: signal: %s isn't supported. Canceling operation.", __FUNCTION__, entry->get_name().c_str());
+						return entry;
+					}
+				}
 		}
 		else
 		{
@@ -473,7 +487,7 @@ active_diagnostic_request_t* diagnostic_manager_t::add_recurring_request(Diagnos
 		}
 	}
 	else
-	{ DEBUG(binder_interface, "%s: Can't add request, one already exists with same key", __FUNCTION__);}
+		{ DEBUG(binder_interface, "%s: Can't add request, one already exists with same key", __FUNCTION__);}
 	return entry;
 }
 
diff --git a/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.hpp b/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.hpp
index 65ff0cb5..66a4d73f 100644
--- a/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.hpp
+++ b/CAN-binder/low-can-binding/diagnostic/diagnostic-manager.hpp
@@ -57,6 +57,7 @@ private:
 																	   * response is received for a non-recurring request or it times out, it is removed*/
 	bool initialized_; /*!< * initialized - True if the DiagnosticsManager has been initialized with shims. It will interface with the uds-c lib*/
 	utils::socketcan_bcm_t socket_; ///< rx_socket_ - a BCM socket with 8 RX_SETUP jobs for the 8 CAN ID on which ECU could respond.
+	struct sd_event_source* event_source_;
 
 	void init_diagnostic_shims();
 	void reset();
@@ -66,6 +67,8 @@ public:
 
 	bool initialize();
 
+	void read_socket();
+	utils::socketcan_bcm_t& get_socket();
 	std::string get_can_bus();
 	active_diagnostic_request_t* get_last_recurring_requests() const;
 	DiagnosticShims& get_shims();