All queue under can_bus_t class and reading thread

function with an additionnal can_bus_t object argument

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

7 files changed, 121 insertions, 123 deletions

diff --git a/src/can-utils.cpp b/src/can-utils.cpp
index a156df4..868eb23 100644
--- a/src/can-utils.cpp
+++ b/src/can-utils.cpp
@@ -232,10 +232,10 @@ canfd_frame can_bus_dev_t::read(const struct afb_binding_interface* interface)
  * @brief start reading threads and set flag is_running_
  * 
  */
-void can_bus_dev_t::start_reading()
+void can_bus_dev_t::start_reading(can_bus_t& can_bus)
 {
-	th_reading_ = std::thread(can_reader, std::ref(*this));
-	is_running_ = true;	is_running_ = true;
+	th_reading_ = std::thread(can_reader, std::ref(*this), std::ref(can_bus));
+	is_running_ = true;
 }
 
 /*
@@ -247,51 +247,6 @@ bool can_bus_dev_t::is_running()
 }
 
 /**
- * @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.
- */
-can_message_t can_bus_dev_t::next_can_message(const struct afb_binding_interface* interface)
-{
-	can_message_t can_msg(interface);
-
-	if(!can_message_q_.empty())
-	{
-		can_msg = can_message_q_.front();
-		can_message_q_.pop();
-		DEBUG(interface, "next_can_message: Here is the next can message : id %d, length %d", can_msg.get_id(), can_msg.get_length());
-		return can_msg;
-	}
-	
-	NOTICE(interface, "next_can_message: End of can message queue");
-	has_can_message_ = false;
-	return can_msg;
-}
-
-/**
- * @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_;
-}
-
-/**
  * @brief Send a can message from a can_message_t object.
  * 
  * params[const can_message_t& can_msg] the can message object to send
@@ -328,8 +283,8 @@ int can_bus_dev_t::send_can_message(can_message_t& can_msg, const struct afb_bin
 *
 *********************************************************************************/
 
-can_bus_t::can_bus_t(const afb_binding_interface *itf, int& conf_file)
-	: interface_{itf}, conf_file_{conf_file}
+can_bus_t::can_bus_t(const struct afb_binding_interface *interface, int& conf_file)
+	: interface_{interface}, conf_file_{conf_file}
 {
 }
 
@@ -366,7 +321,7 @@ int can_bus_t::init_can_dev()
 		{
 			can_bus_dev_t can_bus_device_handler(device);
 			can_bus_device_handler.open(interface_);
-			can_bus_device_handler.start_reading();
+			can_bus_device_handler.start_reading(std::ref(*this));
 			i++;
 		}
 
@@ -421,6 +376,51 @@ std::vector<std::string> can_bus_t::read_conf()
 }
 
 /**
+ * @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.
+ */
+can_message_t can_bus_t::next_can_message()
+{
+	can_message_t can_msg(interface_);
+
+	if(!can_message_q_.empty())
+	{
+		can_msg = can_message_q_.front();
+		can_message_q_.pop();
+		DEBUG(interface_, "next_can_message: Here is the next can message : id %d, length %d", can_msg.get_id(), can_msg.get_length());
+		return can_msg;
+	}
+	
+	NOTICE(interface_, "next_can_message: End of can message queue");
+	has_can_message_ = false;
+	return can_msg;
+}
+
+/**
+ * @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_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_t::has_can_message() const
+{
+	return has_can_message_;
+}
+
+/**
  * @brief: Get a VehicleMessage from vehicle_message_q and return it
  * then point to the next VehicleMessage in queue.
  * 
diff --git a/src/can-utils.hpp b/src/can-utils.hpp
index c6b23a2..c709af4 100644
--- a/src/can-utils.hpp
+++ b/src/can-utils.hpp
@@ -140,41 +140,6 @@ class can_message_t {
 };
 
 /** 
- * @brief Object representing a can device. Handle opening, closing and reading on the
- * socket. This is the low level object to be use by can_bus_t.
- *
- * @params[in] std::string device_name_ - name of the linux device handling the can bus. Generally vcan0, can0, etc.
- */
-class can_bus_dev_t {
-	private:
-		std::string device_name_;
-		int can_socket_;
-		bool is_fdmode_on_;
-		struct sockaddr_can txAddress_;
-
-		bool has_can_message_;
-		std::queue <can_message_t> can_message_q_;
-
-		std::thread th_reading_;
-		bool is_running_;
-
-	public:
-		can_bus_dev_t(const std::string& dev_name);
-
-		int open(const struct afb_binding_interface* interface);
-		int close();
-		bool is_running();
-		void start_reading();
-		canfd_frame read(const struct afb_binding_interface *interface);
-		
-		can_message_t next_can_message(const struct afb_binding_interface* interface);
-		void push_new_can_message(const can_message_t& can_msg);		
-		bool has_can_message() const;
-
-		int send_can_message(can_message_t& can_msg, const struct afb_binding_interface* interface);
-};
-
-/** 
  * @brief Object used to handle decoding and manage event queue to be pushed.
  *
  * @params[in] interface_ - afb_binding_interface pointer to the binder. Used to log messages
@@ -188,6 +153,9 @@ class can_bus_t {
 		std::thread th_decoding_;
 		std::thread th_pushing_;
 
+		bool has_can_message_;
+		std::queue <can_message_t> can_message_q_;
+
 		bool has_vehicle_message_;
 		std::queue <openxc_VehicleMessage> vehicle_message_q_;
 
@@ -197,12 +165,44 @@ class can_bus_t {
 		std::vector<std::string> read_conf();
 		
 		void start_threads();
+
+		can_message_t next_can_message();
+		void push_new_can_message(const can_message_t& can_msg);		
+		bool has_can_message() const;
 		
 		openxc_VehicleMessage next_vehicle_message();
 		void push_new_vehicle_message(const openxc_VehicleMessage& v_msg);
 		bool has_vehicle_message() const;
 };
 
+/** 
+ * @brief Object representing a can device. Handle opening, closing and reading on the
+ * socket. This is the low level object to be use by can_bus_t.
+ *
+ * @params[in] std::string device_name_ - name of the linux device handling the can bus. Generally vcan0, can0, etc.
+ */
+class can_bus_dev_t {
+	private:
+		std::string device_name_;
+		int can_socket_;
+		bool is_fdmode_on_;
+		struct sockaddr_can txAddress_;
+
+		std::thread th_reading_;
+		bool is_running_;
+
+	public:
+		can_bus_dev_t(const std::string& dev_name);
+
+		int open(const struct afb_binding_interface* interface);
+		int close();
+		bool is_running();
+		void start_reading(can_bus_t& can_bus);
+		canfd_frame read(const struct afb_binding_interface *interface);
+		
+		int send_can_message(can_message_t& can_msg, const struct afb_binding_interface* interface);
+};
+
 /**
  * @brief A state encoded (SED) signal's mapping from numerical values to
  * OpenXC state names.
@@ -399,6 +399,6 @@ void logBusStatistics(can_bus_dev_t* buses, const int busCount);
 /**
  * @brief Function representing thread activated by can bus objects
  */
-void can_reader(can_bus_dev_t& can_bus);
+void can_reader(can_bus_dev_t& can_bus_dev, can_bus_t& can_bus);
 void can_decode_message(can_bus_t& can_bus);
 void can_event_push(can_bus_t& can_bus);
\ No newline at end of file
diff --git a/src/can_decode_message.cpp b/src/can_decode_message.cpp
index 46b2977..51993b8 100644
--- a/src/can_decode_message.cpp
+++ b/src/can_decode_message.cpp
@@ -27,31 +27,30 @@ void can_decode_message(can_bus_t &can_bus)
 	openxc_DynamicField search_key, ret;
 	bool send = true;
 
-	decoder_t decoder();
+	decoder_t decoder;
 
-	while(true)
+	while(can_bus.has_can_message())
 	{
-		if(can_message = can_bus.next_can_message(interface))
-		{
-			/* First we have to found which CanSignal is */
-			search_key = build_DynamicField((double)can_message.get_id())
-			signals = find_can_signals(search_key);
+		can_message = can_bus.next_can_message();
+
+		/* First we have to found which CanSignal is */
+		search_key = build_DynamicField((double)can_message.get_id());
+		signals = find_can_signals(search_key);
+		
+		/* Decoding the message ! Don't kill the messenger ! */
+		for(const auto& sig : signals)
+		{	
+			subscribed_signals_i = subscribed_signals.find(sig.genericName);
 			
-			/* Decoding the message ! Don't kill the messenger ! */
-			for(const auto& sig : signals)
-			{	
-				subscribed_signals_i = subscribed_signals.find(sig.genericName);
-				
-				if(subscribed_signals_i != subscribed_signals.end() &&
-					afb_event_is_valid(subscribed_signals_i->second))
-				{
-					ret = decoder.decodeSignal(sig, can_message, getSignals(), &send);
+			if(subscribed_signals_i != subscribed_signals.end() &&
+				afb_event_is_valid(subscribed_signals_i->second))
+			{
+				ret = decoder.decodeSignal(sig, can_message, getSignals(), &send);
+
+				openxc_SimpleMessage s_message = build_SimpleMessage(sig.genericName, ret);
 
-					s_message = build_SimpleMessage(sig.genericName, ret);
-						
-					vehicle_message = build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type::openxc_DynamicField_Type_NUM, s_message);
-					vehicle_message_q.push(vehicle_message);
-				}
+				vehicle_message = build_VehicleMessage_with_SimpleMessage(openxc_DynamicField_Type::openxc_DynamicField_Type_NUM, s_message);
+				can_bus.push_new_vehicle_message(vehicle_message);
 			}
 		}
 	}
diff --git a/src/can_event_push.cpp b/src/can_event_push.cpp
index 9251da6..07ae8d3 100644
--- a/src/can_event_push.cpp
+++ b/src/can_event_push.cpp
@@ -22,16 +22,13 @@ void can_event_push(can_bus_t& can_bus)
 {
 	openxc_VehicleMessage v_message;
 	openxc_SimpleMessage s_message;
-	iterator it_event;
-
-	while(true)
+	
+	while(can_bus.has_vehicle_message())
 	{
-		if(v_message = can_bus->next_vehicle_message())
-		{
-			s_message = get_simple_message(v_msg);
-			const auto& it_event = subscribed_signals.find(s_message.name);
-			if(! it_event->end() && afb_event_is_valid(it_event->second))
-				afb_event_push(it_event->second, jsonify_simple(s_message));
-		}
+		v_message = can_bus.next_vehicle_message();
+		s_message = get_simple_message(v_message);
+		const auto& it_event = subscribed_signals.find(s_message.name);
+		if(it_event != subscribed_signals.end() && afb_event_is_valid(it_event->second))
+			afb_event_push(it_event->second, jsonify_simple(s_message));
 	}
 }
diff --git a/src/can_reader.cpp b/src/can_reader.cpp
index 6fc1e74..4be1546 100644
--- a/src/can_reader.cpp
+++ b/src/can_reader.cpp
@@ -18,13 +18,13 @@
 
 #include "can_reader.hpp"
 
-void can_reader(can_bus_dev_t &can_bus)
+void can_reader(can_bus_dev_t &can_bus_dev, can_bus_t& can_bus)
 {
 	can_message_t can_message(interface);
 
-	while(can_bus.is_running())
+	while(can_bus_dev.is_running())
 	{
-		can_message.convert_from_canfd_frame(can_bus.read(interface));
+		can_message.convert_from_canfd_frame(can_bus_dev.read(interface));
 		can_bus.push_new_can_message(can_message);
 	}
 }
\ No newline at end of file
diff --git a/src/can_reader.hpp b/src/can_reader.hpp
index 813e7b7..2e33897 100644
--- a/src/can_reader.hpp
+++ b/src/can_reader.hpp
@@ -21,4 +21,4 @@
 #include "can-utils.hpp"
 #include "low-can-binding.hpp"
 
-void can_reader(can_bus_dev_t &can_bus);
\ No newline at end of file
+void can_reader(can_bus_dev_t& can_bus_dev, can_bus_t& can_bus);
\ No newline at end of file
diff --git a/src/timer.cpp b/src/timer.cpp
index 70ce2b8..b0ccb62 100644
--- a/src/timer.cpp
+++ b/src/timer.cpp
@@ -14,7 +14,9 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
- 
+
+#include "timer.hpp"
+
 inline unsigned long systemTimeMs()
 {
 	struct timeb t_msec;