Separated can_bus and can_bus_dev object.

Now there are as many thread reading as can bus device

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

3 files changed, 236 insertions, 103 deletions

diff --git a/src/can-utils.cpp b/src/can-utils.cpp
index e41230b..3aa2ca8 100644
--- a/src/can-utils.cpp
+++ b/src/can-utils.cpp
@@ -19,12 +19,13 @@
 
 /********************************************************************************
 *
-*		CanBus method implementation
+*		can_bus_dev_t method implementation
 *
 *********************************************************************************/
 
+
 can_bus_dev_t::can_bus_dev_t(afb_binding_interface *itf, const std:string &dev_name)
-	: interface{itf}, deviceName{dev_name}
+	: device_name_{dev_name}
 {
 }
 
@@ -124,6 +125,16 @@ canfd_frame can_bus_dev_t::read()
 	return canfd_frame;
 }
 
+/**
+ * @brief start reading threads and set flag is_running_
+ * 
+ */
+void can_bus_dev_t::start_reading()
+{
+	th_reading_ = std::thread(can_reader, this);
+	is_running_ = true;
+}
+
 /*
  * Return is_running_ bool
  */
@@ -132,25 +143,11 @@ bool can_bus_dev_t::is_running()
 	return is_running_;
 }
 
-can_bus_t::can_bus_t(afb_binding_interface *itf, const std:string &dev_name)
-	: interface{itf}
-{
-}
-
-void can_bus_t::start_threads()
-{
-	th_reading_ = std::thread(can_reader, interface, socket, can_message_q_);
-	is_running_ = true;
-
-	th_decoding_ = std::thread(can_decoder, interface, can_message_q, can_message_q_);
-	th_pushing_ = std::thread(can_event_push, interface, can_message_q_);
-}
-
-/*
- * Get a CanMessage from can_message_q and return it
- * then point to the next CanMessage in queue.
+/**
+ * @brief: Get a can_message_t from can_message_q and return it
+ * then point to the next can_message_t in queue.
  * 
- * Return the next queue element or NULL if queue is empty.
+ * @return the next queue element or NULL if queue is empty.
  */
 can_message_t can_bus_dev_t::next_can_message()
 {
@@ -160,26 +157,134 @@ can_message_t can_bus_dev_t::next_can_message()
 		can_message_q_.pop()
 		return &can_msg;
 	}
+	
 	has_can_message_ = false;
 }
 
 /**
- * @return has_can_message_ bool
+ * @brief Append a new element to the can message queue and set
+ * has_can_message_ boolean to true
+ * 
+ * @params[const can_message_t& can_msg] the can_message_t to append
+ * 
+ */
+void can_bus_dev_t::push_new_can_message(const can_message_t& can_msg)
+{
+	can_message_q_.push(can_msg);
+}
+
+/**
+ * @brief Flag that let you know when can message queue is exhausted
+ * 
+ * @return[bool] has_can_message_ bool
  */
 bool can_bus_dev_t::has_can_message() const
 {
 	return has_can_message_;
 }
 
-void can_bus_dev_t::push_new_can_message(const can_message_t& can_msg)
+/********************************************************************************
+*
+*		can_bus_t method implementation
+*
+*********************************************************************************/
+
+can_bus_t::can_bus_t(afb_binding_interface *itf, std::ifstream& conf_file)
+	: interface{itf}, conf_file_{conf_file}
 {
-	can_message_q_.push(can_msg);
 }
 
-/*
- * Send a can message from a can_message_t object.
+/**
+ * @brief start threads relative to the can bus: decoding and pushing
+ * as the reading is handled by can_bus_dev_t object
+ * 
+ */
+void can_bus_t::start_threads()
+{
+	th_decoding_ = std::thread(can_decoder, this);
+	th_pushing_ = std::thread(can_event_push, this);
+}
+
+
+/**
+ * @brief Initialize as many as can_bus_dev_t objects with their respective reading thread
+ * 
+ * params[std::ifstream& conf_file] conf_file ifstream to the JSON configuration 
+ * file located at the rootdir of the binding
+ */
+ void init_can_dev()
+ {
+   std::vector<std::string> devices_name;
+   int i, t;
+   
+   devices_name = read_conf(conf_file_);
+   
+   t = devices_name.size();
+   i=0
+   
+   for(const auto& device : devices_name)
+   {
+     can_bus_dev_t(device);
+     i++;
+   }
+   
+   NOTICE(interface_, "Initialized %d/%d can bus device(s)", i, t);
+ }
+
+/** 
+ * @brief Read the conf file and extract device name
+ * 
+ * @params[std::ifstream& conf_file] conf_file JSON configuration
+ * file located at the rootdir of the binding
+ * 
+ * @return[std:vector<std::string>] return a vector of device name
  */
-int can_bus_t::send_can_message(can_message_t &can_msg)
+ std::vector<std::string> read_conf(std::ifstream& conf_file)
+ {
+  std::vector<std::string> ret;
+  std::string fd_conf_content;
+	json_object jo, canbus;
+  int n, i, ok;
+  
+	/* Open JSON conf file */
+	if (conf_file)
+	{
+		conf_file.seekg(0, std::ios::end);
+		conf_file.resize(conf_file.tellg());
+		conf_file.seekg(0, std::ios::beg);
+		conf_file.read(&fd_conf_content[0], fd_conf_content.size());
+		conf_file.close();
+
+  	jo = json_tokener_parse(&fd_conf_content);
+  
+    if (jo == NULL || !json_object_object_get_ex(&jo, "canbus", &&canbus))
+      ERROR(interface_, "Can't find canbus node in the configuration file. Please review it.");
+    else if (json_object_get_type(canbus) != json_type_array)
+  		ret.push_back(json_object_get_string(a));
+  	else
+  	{
+  		n = json_object_array_length(a);
+  		ok = 0;
+  		for (i = 0 ; i < n ; i++)
+  			ret.push_back(json_object_get_string(json_object_array_get_idx(a, i)));
+    }
+    return ret;
+	}
+  else
+  {
+    ERROR(interface_, "Problem at reading the conf file");
+    return 0;
+  }
+}
+
+/**
+ * @brief Send a can message from a can_message_t object.
+ * TODO: specify which can_dev to use as we can use many
+ * 
+ * params[const can_message_t& can_msg] the can message object to send
+ * 
+ */
+int can_bus_t::send_can_message(const can_message_t &can_msg)
 {
 	int nbytes;
 	canfd_frame *f;
@@ -206,11 +311,11 @@ int can_bus_t::send_can_message(can_message_t &can_msg)
 	return 0;
 }
 
-/*
- * Get a VehicleMessage from vehicle_message_q and return it
+/**
+ * @brief: Get a VehicleMessage from vehicle_message_q and return it
  * then point to the next VehicleMessage in queue.
  * 
- * Return the next queue element or NULL if queue is empty.
+ * @return the next queue element or NULL if queue is empty.
  */
 openxc_VehicleMessage* can_bus_t::next_vehicle_message()
 {
@@ -224,14 +329,23 @@ openxc_VehicleMessage* can_bus_t::next_vehicle_message()
 	has_vehicle_message_ = false;
 }
 
+/**
+ * @brief Append a new element to the vehicle message queue and set
+ * has_vehicle_message_ boolean to true
+ * 
+ * @params[const openxc_VehicleMessage& v_msg] the openxc_VehicleMessage to append
+ * 
+ */
 void can_bus_t::push_new_vehicle_message(const openxc_VehicleMessage& v_msg)
 {
 	vehicle_message_q_.push(v_msg);
 	has_vehicle_message_ = true;
 }
 
-/*
- * Return has_can_message_ bool
+/**
+ * @brief Flag that let you know when vehicle message queue is exhausted
+ * 
+ * @return[bool] has_vehicle_message_ bool
  */
 bool can_bus_t::has_vehicle_message() const
 {
diff --git a/src/can-utils.hpp b/src/can-utils.hpp
index da5f81c..8b64cb3 100644
--- a/src/can-utils.hpp
+++ b/src/can-utils.hpp
@@ -18,7 +18,7 @@
 #pragma once
 
 #include <string>
-#include "timer.h"
+#include "timer.hpp"
 #include "openxc.pb.h"
 
 // TODO actual max is 32 but dropped to 24 for memory considerations
@@ -74,8 +74,6 @@ class can_bus_dev_t {
 	private:
 		std::string device_name_;
 		int can_socket_;
-
-		int can_socket_;
 		bool is_fdmode_on_;
 		struct sockaddr_can txAddress_;
 
@@ -92,6 +90,7 @@ class can_bus_dev_t {
 		
 		can_message_t* next_can_message();
 		void push_new_can_message(const can_message_t& can_msg);		
+		bool has_can_message() const;
 }
 
 
@@ -105,9 +104,7 @@ class can_bus_dev_t {
 class can_bus_t {
 	private:
 		afb_binding_interface *interface_;
-
-		std::vector<can_bus_dev_t> devices;
-
+		
 		std::thread th_decoding_;
 		std::thread th_pushing_;
 
@@ -116,11 +113,15 @@ class can_bus_t {
 
 	public:
 		void start_threads();
+		
+		void init_can_dev(std::ifstream& conf_file);
+		std::vector<std::string> read_conf()
 
 		int send_can_message(can_message_t can_msg);
 
 		openxc_VehicleMessage& next_vehicle_message();
 		void push_new_vehicle_message(const openxc_VehicleMessage& v_msg);
+		bool has_vehicle_message() const;
 };
 
 /* A compact representation of a single CAN message, meant to be used in in/out
diff --git a/src/low-can-binding.cpp b/src/low-can-binding.cpp
index b895adc..7899797 100644
--- a/src/low-can-binding.cpp
+++ b/src/low-can-binding.cpp
@@ -40,7 +40,9 @@
 #include <afb/afb-binding.h>
 #include <afb/afb-service-itf.h>
 
-#include "obd2.hpp"
+//#include "obd2.hpp"
+#include "can-utils.hpp"
+#include "can-signals.hpp"
 
 /*
  *	 Interface between the daemon and the binding
@@ -56,21 +58,22 @@ static const struct afb_binding_interface *interface;
 /*
  * TBF TBF TBF
  * called on an event on the CAN bus
- */
-static int on_event(sd_event_source *s, int fd, uint32_t revents, void *userdata)
+ static int on_event(sd_event_source *s, int fd, uint32_t revents, void *userdata)
 {
 	openxc_CanMessage can_message;
 
 	can_message = openxc_CanMessage_init_default;
 
 	/* read available data */
+	/*
 	if ((revents & EPOLLIN) != 0)
 	{
 		read_can(&can_message);
 		send_event();
 	}
-
+*/
 	/* check if error or hangup */
+/*
 	if ((revents & (EPOLLERR|EPOLLRDHUP|EPOLLHUP)) != 0)
 	{
 		sd_event_source_unref(s);
@@ -80,7 +83,7 @@ static int on_event(sd_event_source *s, int fd, uint32_t revents, void *userdata
 
 	return 0;
 }
-
+*/
 /*
  * USELESS SINCE THREADS SEPARATION
  *
@@ -120,45 +123,68 @@ static int connect_to_event_loop(CanBus &CanBus_handler)
 *
 *********************************************************************************/
 
-static int subscribe_unsubscribe_signal(struct afb_req request, int subscribe, std::map<CanSignal, struct afb_event>::iterator *s_sig)
+static int subscribe_unsubscribe_signal(struct afb_req request, bool subscribe, std::vector<CanSignal>::const_iterator& sig_i)
  {
-	if (!afb_event_is_valid(s_sig->second)) {
-		if (!subscribe)
-			return 1;
-		sig->event = afb_daemon_make_event(afbitf->daemon, s_sig->first->genericName);
-		if (!afb_event_is_valid(s_sig->second)) {
-			return 0;
-		}
-	}
-
-	if (((subscribe ? afb_req_subscribe : afb_req_unsubscribe)(request, s_sig->second)) < 0) {
-		return 0;
+  int ret;
+  
+  const auto& ss_i = subscribed_signals.find(sig_i);
+  if (ss_i != subscribed_signals.end())
+  {
+    if(!afb_event_is_valid(ss_i->second))
+    {
+      if(!subscribe)
+      {
+        NOTICE(interface, "Event isn't valid, it can't be unsubscribed.");
+        ret = 1;
+      }
+      else
+      {
+        ss_i->second = afb_daemon_make_event(afbitf->daemon, ss_i->first.genericName);
+        if (!afb_event_is_valid(ss_i->second)) 
+        {
+          ERROR(interface, "Can't create an event, something goes wrong.");
+          ret = 0;
+        }
+      }
+    }
+  }
+  else
+  {
+    subscribed_signals[sig_i] = afb_daemon_make_event(afbitf->daemon, sig_i.genericName);
+    if (!afb_event_is_valid(ss_i->second)) 
+    {
+      ERROR(interface, "Can't create an event, something goes wrong.");
+      ret = 0;
+    }
+ }
+          
+	if (((subscribe ? afb_req_subscribe : afb_req_unsubscribe)(request, subscribed_signals[sig_i])) < 0)
+	{
+	  ERROR(interface, "Operation goes wrong for signal: %s", sig_i.genericName);
+	  ret = 0;
 	}
-
-	return 1;
+  else
+	  ret = 1;
+	
+	return ret;
  }
 
-static int subscribe_unsubscribe_signals(struct afb_req request, int subscribe, std:vector<CanSignal> *signals)
+static int subscribe_unsubscribe_signals(struct afb_req request, bool subscribe, const std:vector<CanSignal>& signals)
 {
 	std::vector<CanSignal>::iterator signal_i;
 	std::map <CanSignal, struct afb_event>::iterator s_signal_i;
-	
-	for(signal_i=signals.begin(); signal_i != signals.end(); signal_i++)
+  int ret;
+
+	// TODO: lock the subscribed_signals when insert/remove
+	for(const auto& signal_i : signals)
 	{
-		s_signal_i = subscribed_signals.find(signal_i);
-		if(s_signal_i != subscribed_signals.end())
-			subscribe_unsubscribe_signal(request, subscribe, s_signal_i);
-		else
-		{
-			std::map <CanSignal, struct afb_event>::iterator it = subscribed_signals.begin();
-			it = subscribed_signals.insert(it, std::pair<CanSignal, struct afb_event>(signal_i, NULL);
-			subscribe_unsubscribe_signal(request, subscribe, it);
-		}
-		return 0;
+	  ret = subscribe_unsubscribe_signal(request, subscribe, signal_i);
+	  if(ret == 0)
+	    return ret;
 	}
 }
 
-static int subscribe_unsubscribe_all(struct afb_req request, int subscribe)
+static int subscribe_unsubscribe_all(struct afb_req request, bool subscribe)
 {
 	int i, n, e;
 
@@ -169,25 +195,32 @@ static int subscribe_unsubscribe_all(struct afb_req request, int subscribe)
 	return e == 0;
 }
 
-static int subscribe_unsubscribe_name(struct afb_req request, int subscribe, const char *name)
+static int subscribe_unsubscribe_name(struct afb_req request, bool subscribe, const char *name)
 {
-	 std::vector <CanSignal> *sig;
-
-	if (0 == strcmp(name, "*"))
-		return subscribe_unsubscribe_all(request, subscribe);
-
-	if(obd2_handler_c.is_obd2_signal(name))
+	 std::vector<CanSignal> sig;
+	 int ret = 0;
 
+	if (!strcmp(name, "*"))
+	  ret = subscribe_unsubscribe_all(request, subscribe);
 	else
-		find_can_signals(name, sig);
-	if (sig == NULL) {
-		return 0;
-	}
-
-	return subscribe_unsubscribe_sig(request, subscribe, sig);
+	{
+		//if(obd2_handler_c.is_obd2_signal(name))
+		if(false)
+		{
+		// TODO
+		}
+		else
+		{
+		sig = find_can_signals(name);
+		if (sig.empty())
+			ret = 0;
+  	}
+  	ret = subscribe_unsubscribe_signals(request, subscribe, sig);
+  }
+	return ret;
 }
 
-static void subscribe_unsubscribe(struct afb_req request, int subscribe)
+static void subscribe_unsubscribe(struct afb_req request, bool subscribe)
 {
 	int ok, i, n;
 	struct json_object *args, *a, *x;
@@ -218,12 +251,12 @@ static void subscribe_unsubscribe(struct afb_req request, int subscribe)
 
 static void subscribe(struct afb_req request)
 {
-	subscribe_unsubscribe(request, 1);
+	subscribe_unsubscribe(request, true);
 }
 
 static void unsubscribe(struct afb_req request)
 {
-	subscribe_unsubscribe(request, 0);
+	subscribe_unsubscribe(request, false);
 }
 static const struct afb_verb_desc_v1 verbs[]=
 {
@@ -258,25 +291,10 @@ const struct afb_binding *afbBindingV1Register (const struct afb_binding_interfa
 int afbBindingV1ServiceInit(struct afb_service service)
 {
 	std::ifstream fd_conf;
-	std::string fd_conf_content;
-	json_object jo_canbus;
-
-	/* Open JSON conf file */
-	jo_canbus = json_object_new_object();
-	fd_conf = afb_daemon_rootdir_open_locale(interface->daemon, "canbus.json", O_RDONLY, NULL);
-	if (fd_conf)
-	{
-		fd_conf.seekg(0, std::ios::end);
-		fd_conf_content.resize(fd_conf.tellg());
-		fd_conf.seekg(0, std::ios::beg);
-		fd_conf.read(&fd_conf_content[0], fd_conf_content.size());
-		fd_conf.close();
-	}
-
-	jo_canbus = json_tokener_parse(&fd_conf_content);
+	fd_conf = afb_daemon_rootdir_open_locale(interface->daemon, "can_bus.json", O_RDONLY, NULL);
 
 	/* Open CAN socket */
-	can_bus_t CanBus_handler(interface, json_object_get_string(json_object_object_get(jo_canbus, "deviceName"));
+	can_bus_t can_bus_handler(interface, ));
 	CanBus_handler.open();
 	CanBus_handler.start_threads();
 }