Make and use a single function to read incoming CAN messages

[apps/low-level-can-service.git] / CAN-binder / low-can-binding / binding / low-can-cb.cpp

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 df6795f..4faae85 100644 (file)
--- a/CAN-binder/low-can-binding/binding/low-can-cb.cpp
+++ b/CAN-binder/low-can-binding/binding/low-can-cb.cpp
@@ -52,21 +52,57 @@ void on_no_clients(std::string message)
        }
 }
 
-int read(sd_event_source *s, int fd, uint32_t revents, void *userdata)
+static void push_n_notify(const can_message_t cm)
 {
-       can_signal_t* sig= (can_signal_t*)userdata;
-       sig->read_socket();
+       can_bus_t& cbm = configuration_t::instance().get_can_bus_manager();
+       std::lock_guard<std::mutex> can_message_lock(cbm.get_can_message_mutex());
+       { cbm.push_new_can_message(cm); }
+       cbm.get_new_can_message_cv().notify_one();
+}
+
+int read_message(sd_event_source *s, int fd, uint32_t revents, void *userdata)
+{
+       can_message_t cm;
+       can_signal_t* sig;
+       diagnostic_manager_t& diag_m = configuration_t::instance().get_diagnostic_manager();
+
+       if(userdata != nullptr)
+       {
+               sig = (can_signal_t*)userdata;
+               utils::socketcan_bcm_t s = sig->get_socket();
+               s >> cm;
+       }
+       else
+       {
+               utils::socketcan_bcm_t s = diag_m.get_socket();
+               s >> cm;
+       }
+
+       push_n_notify(cm);
 
        /* check if error or hangup */
        if ((revents & (EPOLLERR|EPOLLRDHUP|EPOLLHUP)) != 0)
        {
                sd_event_source_unref(s);
-               sig->get_socket().close();
-               sig->create_rx_filter();
+               if(userdata != nullptr)
+               {
+                       sig->get_socket().close();
+                       sig->create_rx_filter();
+                       NOTICE(binder_interface, "%s: Recreated RX_SETUP BCM job for signal: %s", __FUNCTION__, sig->get_name().c_str());
+               }
+               else
+               {
+                       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 +197,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
                {
@@ -189,14 +224,14 @@ 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_message, sig.get());
                rets++;
                DEBUG(binder_interface, "%s: signal: %s subscribed", __FUNCTION__, sig->get_name().c_str());
        }
        return rets;
 }
 
-static int process_args(struct afb_req request, std::vector<std::string> args, bool subscribe)
+static int process_args(struct afb_req request, const std::vector<std::string>& args, bool subscribe)
 {
        struct utils::signals_found sf;
        int ok = 0, total = 0;
@@ -216,7 +251,7 @@ static int process_args(struct afb_req request, std::vector<std::string> args, b
        return ok;
 }
 
-static const std::vector<std::string> parse_args_from_request(struct afb_req request, bool subscribe)
+static const std::vector<std::string> parse_args_from_request(struct afb_req request)
 {
        int i, n;
        std::vector<std::string> ret;
@@ -247,10 +282,9 @@ static const std::vector<std::string> parse_args_from_request(struct afb_req req
 
 void subscribe(struct afb_req request)
 {
-       std::vector<std::string> args;
        bool subscribe = true;
 
-       args = parse_args_from_request(request, subscribe);
+       const std::vector<std::string> args = parse_args_from_request(request);
 
        if (process_args(request, args, subscribe) > 0)
                afb_req_success(request, NULL, NULL);
@@ -263,7 +297,7 @@ void unsubscribe(struct afb_req request)
        std::vector<std::string> args;
        bool subscribe = false;
        
-       args = parse_args_from_request(request, subscribe);
+       args = parse_args_from_request(request);
 
        if (process_args(request, args, subscribe) > 0)
                afb_req_success(request, NULL, NULL);