fix(rtsp): Fix issue preventing client destructor from completing (#463)

* fix(rtsp): Fix issue preventing client destructor from completing

* ensure all rtsp sockets that are used for receive have proper receive timeouts. update logging for expected error conditions to be info instead

* make timeouts configurable

* add missing slash

Author: finger563

components/rtsp/include/rtsp_client.hpp

@@ -98,6 +98,7 @@ class RtspClient : public BaseComponent {
   /// \note Starts the RTP and RTCP threads.
   /// Sends the SETUP request to the RTSP server and parses the response.
   /// \note The default ports are 5000 and 5001 for RTP and RTCP respectively.
+  /// \note The default receive timeout is 5 seconds.
   /// \param ec The error code to set if an error occurs
   void setup(std::error_code &ec);
 
@@ -106,8 +107,10 @@ class RtspClient : public BaseComponent {
   /// \note Starts the RTP and RTCP threads.
   /// \param rtp_port The RTP client port
   /// \param rtcp_port The RTCP client port
+  /// \param receive_timeout The timeout for receiving RTP and RTCP packets
   /// \param ec The error code to set if an error occurs
-  void setup(size_t rtp_port, size_t rtcp_port, std::error_code &ec);
+  void setup(size_t rtp_port, size_t rtcp_port, const std::chrono::duration<float> &receive_timeout,
+             std::error_code &ec);
 
   /// Play the RTSP stream
   /// Sends the PLAY request to the RTSP server and parses the response.
@@ -139,14 +142,18 @@ class RtspClient : public BaseComponent {
   /// Initialize the RTP socket
   /// \note Starts the RTP socket task.
   /// \param rtp_port The RTP client port
+  /// \param receive_timeout The timeout for receiving RTP packets
   /// \param ec The error code to set if an error occurs
-  void init_rtp(size_t rtp_port, std::error_code &ec);
+  void init_rtp(size_t rtp_port, const std::chrono::duration<float> &receive_timeout,
+                std::error_code &ec);
 
   /// Initialize the RTCP socket
   /// \note Starts the RTCP socket task.
   /// \param rtcp_port The RTCP client port
+  /// \param receive_timeout The timeout for receiving RTCP packets
   /// \param ec The error code to set if an error occurs
-  void init_rtcp(size_t rtcp_port, std::error_code &ec);
+  void init_rtcp(size_t rtcp_port, const std::chrono::duration<float> &receive_timeout,
+                 std::error_code &ec);
 
   /// Handle an RTP packet
   /// \note Parses the RTP packet and appends it to the current JPEG frame.

components/rtsp/include/rtsp_server.hpp

@@ -66,8 +66,9 @@ class RtspServer : public BaseComponent {
 
   /// @brief Start the RTSP server
   /// Starts the accept task, session task, and binds the RTSP socket
+  /// @param accept_timeout The timeout for accepting new connections
   /// @return True if the server was started successfully, false otherwise
-  bool start();
+  bool start(const std::chrono::duration<float> &accept_timeout = std::chrono::seconds(5));
 
   /// @brief Stop the FTP server
   /// Stops the accept task, session task, and closes the RTSP socket

components/rtsp/include/rtsp_session.hpp

@@ -30,6 +30,8 @@ class RtspSession : public BaseComponent {
   struct Config {
     std::string server_address; ///< The address of the server
     std::string rtsp_path;      ///< The RTSP path of the session
+    std::chrono::duration<float> receive_timeout =
+        std::chrono::seconds(5); ///< The timeout for receiving data. Should be > 0.
     espp::Logger::Verbosity log_level =
         espp::Logger::Verbosity::WARN; ///< The log level of the session
   };

components/rtsp/src/rtsp_client.cpp

@@ -150,10 +150,15 @@ void RtspClient::describe(std::error_code &ec) {
 
 void RtspClient::setup(std::error_code &ec) {
   // default to rtp and rtcp client ports 5000 and 5001
-  setup(5000, 50001, ec);
+  using namespace std::chrono_literals;
+  static constexpr size_t rtp_port = 5000;
+  static constexpr size_t rtcp_port = 5001;
+  static constexpr auto receive_timeout = 5s;
+  setup(rtp_port, rtcp_port, receive_timeout, ec);
 }
 
-void RtspClient::setup(size_t rtp_port, size_t rtcp_port, std::error_code &ec) {
+void RtspClient::setup(size_t rtp_port, size_t rtcp_port,
+                       const std::chrono::duration<float> &receive_timeout, std::error_code &ec) {
   // exit early if the error code is set
   if (ec) {
     return;
@@ -169,8 +174,8 @@ void RtspClient::setup(size_t rtp_port, size_t rtcp_port, std::error_code &ec) {
     return;
   }
 
-  init_rtp(rtp_port, ec);
-  init_rtcp(rtcp_port, ec);
+  init_rtp(rtp_port, receive_timeout, ec);
+  init_rtcp(rtcp_port, receive_timeout, ec);
 }
 
 void RtspClient::play(std::error_code &ec) {
@@ -238,12 +243,14 @@ bool RtspClient::parse_response(const std::string &response_data, std::error_cod
   return true;
 }
 
-void RtspClient::init_rtp(size_t rtp_port, std::error_code &ec) {
+void RtspClient::init_rtp(size_t rtp_port, const std::chrono::duration<float> &receive_timeout,
+                          std::error_code &ec) {
   // exit early if the error code is set
   if (ec) {
     return;
   }
   logger_.debug("Starting rtp socket");
+  rtp_socket_.set_receive_timeout(receive_timeout);
   auto rtp_task_config = espp::Task::BaseConfig{
       .name = "Rtp",
       .stack_size_bytes = 16 * 1024,
@@ -261,12 +268,14 @@ void RtspClient::init_rtp(size_t rtp_port, std::error_code &ec) {
   }
 }
 
-void RtspClient::init_rtcp(size_t rtcp_port, std::error_code &ec) {
+void RtspClient::init_rtcp(size_t rtcp_port, const std::chrono::duration<float> &receive_timeout,
+                           std::error_code &ec) {
   // exit early if the error code is set
   if (ec) {
     return;
   }
   logger_.debug("Starting rtcp socket");
+  rtcp_socket_.set_receive_timeout(receive_timeout);
   auto rtcp_task_config = espp::Task::BaseConfig{
       .name = "Rtcp",
       .stack_size_bytes = 6 * 1024,

components/rtsp/src/rtsp_server.cpp

@@ -26,14 +26,18 @@ void RtspServer::set_session_log_level(Logger::Verbosity log_level) {
   session_log_level_ = log_level;
 }
 
-bool RtspServer::start() {
+bool RtspServer::start(const std::chrono::duration<float> &accept_timeout) {
   if (accept_task_ && accept_task_->is_started()) {
     logger_.error("Server is already running");
     return false;
   }
 
   logger_.info("Starting RTSP server on port {}", port_);
 
+  // ensure the receive timeout is set so that the accept will not block
+  // indefinitely and the accept task can be stopped.
+  rtsp_socket_.set_receive_timeout(accept_timeout);
+
   if (!rtsp_socket_.bind(port_)) {
     logger_.error("Failed to bind to port {}", port_);
     return false;
@@ -161,7 +165,13 @@ bool RtspServer::accept_task_function(std::mutex &m, std::condition_variable &cv
   // accept a new connection
   auto control_socket = rtsp_socket_.accept();
   if (!control_socket) {
-    logger_.error("Failed to accept new connection");
+    logger_.info("Failed to accept new connection");
+    // if we were notified, then we should stop the task
+    if (task_notified) {
+      task_notified = false;
+      return true;
+    }
+    // do not stop the task, just try to accept another connection
     return false;
   }
 
@@ -194,7 +204,7 @@ bool RtspServer::accept_task_function(std::mutex &m, std::condition_variable &cv
     session_task_->start();
   }
   // we do not want to stop the task
-  return false;
+  return task_notified;
 }
 
 bool RtspServer::session_task_function(std::mutex &m, std::condition_variable &cv,
@@ -203,8 +213,11 @@ bool RtspServer::session_task_function(std::mutex &m, std::condition_variable &c
   {
     using namespace std::chrono_literals;
     std::unique_lock<std::mutex> lk(m);
-    cv.wait_for(lk, 10ms, [&task_notified] { return task_notified; });
+    auto stop_requested = cv.wait_for(lk, 10ms, [&task_notified] { return task_notified; });
     task_notified = false;
+    if (stop_requested) {
+      return true;
+    }
   }
 
   // when this function returns, the vector of pointers will go out of scope

components/rtsp/src/rtsp_session.cpp

@@ -13,6 +13,8 @@ RtspSession::RtspSession(std::unique_ptr<TcpSocket> control_socket, const Config
     , client_address_(control_socket_->get_remote_info().address) {
   // set the logger tag to include the session id
   logger_.set_tag("RtspSession " + std::to_string(session_id_));
+  // ensure there is a timeout on the control socket receive
+  control_socket_->set_receive_timeout(config.receive_timeout);
   // start the session task to handle RTSP commands
   using namespace std::placeholders;
   control_task_ = std::make_unique<Task>(Task::Config{
@@ -296,6 +298,16 @@ bool RtspSession::control_task_fn(std::mutex &m, std::condition_variable &cv, bo
     if (!handle_rtsp_request(request)) {
       logger_.warn("Failed to handle RTSP request");
     }
+  } else {
+    // if the receive failed, then let's wait a little / check the task_notified
+    // flag to know if we should stop or not.
+    using namespace std::chrono_literals;
+    std::unique_lock<std::mutex> lk(m);
+    auto stop_requested = cv.wait_for(lk, 1ms, [&task_notified] { return task_notified; });
+    task_notified = false;
+    if (stop_requested) {
+      return true;
+    }
   }
   // the receive handles most of the blocking, so we don't need to sleep
   // here, just return false to keep the task running

components/socket/src/tcp_socket.cpp

@@ -185,7 +185,7 @@ std::unique_ptr<TcpSocket> TcpSocket::accept() {
   // accept connection
   auto accepted_socket = ::accept(socket_, (struct sockaddr *)sender_address, &socklen);
   if (accepted_socket < 0) {
-    logger_.error("Could not accept connection: {}", error_string());
+    logger_.info("Could not accept connection: {}", error_string());
     return nullptr;
   }
   connected_client_info.update();

components/socket/src/udp_socket.cpp

@@ -90,7 +90,7 @@ bool UdpSocket::receive(size_t max_num_bytes, std::vector<uint8_t> &data,
                                     (struct sockaddr *)remote_address, &socklen);
   // if we didn't receive anything return false and don't do anything else
   if (num_bytes_received < 0) {
-    logger_.error("Receive failed: {}", error_string());
+    logger_.info("Receive failed: {}", error_string());
     return false;
   }
   // we received data, so call the callback function if one was provided.
@@ -157,8 +157,11 @@ bool UdpSocket::server_task_function(size_t buffer_size, std::mutex &m, std::con
     // if we failed to receive, then likely we should delay a little bit
     using namespace std::chrono_literals;
     std::unique_lock<std::mutex> lk(m);
-    cv.wait_for(lk, 1ms, [&task_notified] { return task_notified; });
+    auto stop_requested = cv.wait_for(lk, 1ms, [&task_notified] { return task_notified; });
     task_notified = false;
+    if (stop_requested) {
+      return true;
+    }
     return false;
   }
   if (!server_receive_callback_) {