X-Git-Url: https://gerrit.automotivelinux.org/gerrit/gitweb?a=blobdiff_plain;f=CAN-binder%2Flow-can-binding%2Fcan%2Fcan-bus.cpp;h=5f45624df4ceb8f53161c9acf435c409aee89b5a;hb=12e680a3c97a2750c657a8c561a79706f3689149;hp=6e74b86af9a4ede1309d8b6d0ccca807c54eab80;hpb=14043c56960ced9141ef107c66a6a041a75f6cb3;p=apps%2Fagl-service-can-low-level.git diff --git a/CAN-binder/low-can-binding/can/can-bus.cpp b/CAN-binder/low-can-binding/can/can-bus.cpp index 6e74b86a..5f45624d 100644 --- a/CAN-binder/low-can-binding/can/can-bus.cpp +++ b/CAN-binder/low-can-binding/can/can-bus.cpp @@ -47,6 +47,51 @@ can_bus_t::can_bus_t(utils::config_parser_t conf_file) std::map> can_bus_t::can_devices_; +/// @brief Listen for all device sockets and fill can_messages_queue with them. +/// Reading blocks until message arrive on listened sockets. +/// +/// @return 0 if ok -1 if not +int can_bus_t::can_reader() +{ + fd_set rfds; + int sock_max = INVALID_SOCKET; + + FD_ZERO(&rfds); + + for(auto& can_dev : can_devices_) + { + FD_SET(can_dev.second->get_socket().socket(), &rfds); + if (sock_max < can_dev.second->get_socket().socket()) + sock_max = can_dev.second->get_socket().socket(); + } + + int ret; + while(is_reading_) + { + ret = select(sock_max + 1, &rfds, nullptr, nullptr, nullptr); + + if(ret == -1) + perror("select()"); + else if(ret > 0) + { + for(const auto& s: can_devices_) + { + if(FD_ISSET(s.second->get_socket().socket(), &rfds)) + { + can_message_t msg; + s.second->get_socket() >> msg; + std::lock_guard can_message_lock(get_can_message_mutex()); + { push_new_can_message(msg); } + get_new_can_message_cv().notify_one(); + } + } + } + else + printf("Timeout\n"); + } + return 0; +} + /// @brief Will make the decoding operation on a classic CAN message. It will not /// handle CAN commands nor diagnostic messages that have their own method to get /// this happens. @@ -83,13 +128,17 @@ int can_bus_t::process_can_signals(can_message_t& can_message) //DEBUG(binder_interface, "Nb elt matched string: %d", (int)s.count(std::string(sig.generic_name)); if( s.find(sig->get_name()) != s.end() && afb_event_is_valid(s[sig->get_name()])) { - decoded_message = decoder_t::translateSignal(*sig, can_message, conf.get_can_signals()); + bool send = true; + decoded_message = decoder_t::translateSignal(*sig, can_message, conf.get_can_signals(), &send); - openxc_SimpleMessage s_message = build_SimpleMessage(sig->get_name(), decoded_message); - vehicle_message = build_VehicleMessage(s_message); + if(send) + { + openxc_SimpleMessage s_message = build_SimpleMessage(sig->get_name(), decoded_message); + vehicle_message = build_VehicleMessage(s_message); - std::lock_guard decoded_can_message_lock(decoded_can_message_mutex_); - push_new_vehicle_message(vehicle_message); + std::lock_guard decoded_can_message_lock(decoded_can_message_mutex_); + push_new_vehicle_message(vehicle_message); + } processed_signals++; } } @@ -199,6 +248,11 @@ void can_bus_t::can_event_push() /// and push subscribed events. void can_bus_t::start_threads() { + is_reading_ = true; + th_reading_ = std::thread(&can_bus_t::can_reader, this); + if(!th_reading_.joinable()) + is_reading_ = false; + is_decoding_ = true; th_decoding_ = std::thread(&can_bus_t::can_decode_message, this); if(!th_decoding_.joinable()) @@ -215,6 +269,7 @@ void can_bus_t::start_threads() /// they'll finish their job. void can_bus_t::stop_threads() { + is_reading_ = false; is_decoding_ = false; is_pushing_ = false; } @@ -248,7 +303,7 @@ int can_bus_t::init_can_dev() can_bus_t::can_devices_[device]->configure(); DEBUG(binder_interface, "%s: Start reading thread", __FUNCTION__); NOTICE(binder_interface, "%s: %s device opened and reading", __FUNCTION__, device.c_str()); - can_bus_t::can_devices_[device]->start_reading(*this); + //can_bus_t::can_devices_[device]->start_reading(*this); i++; } else