Feature/#50

CMakeLists.txt

@@ -1,6 +1,6 @@
 cmake_minimum_required(VERSION 3.16)
 
-project (libflute VERSION 0.12.3)
+project (libflute VERSION 0.12.4)
 
 include(CheckCXXSymbolExists)
 

include/Transmitter.h

@@ -17,6 +17,7 @@
 #pragma once
 #include <boost/asio.hpp>
 #include <boost/bind/bind.hpp>
+#include <atomic>
 #include <chrono>
 #include <queue>
 #include <string>
@@ -616,17 +617,27 @@ namespace LibFlute {
       */
       void activate();
 
-     /**
-      * Deactivate the FLUTE session
-      *
-      * If the Transmitter is currently active then the FLUTE stream is halted and the state is changed to deactivated. Sending of
-      * packets will be halted until the activate() method is called. Note that this will pause File transmission part way through
-      * if a File is currently being transmitted. If the application wishes for deactivation once Files have finished sending then
-      * it should only deactivate() when the completion callback is called and number_of_files() equals 0 to ensure all Files have
-      * been completely transmitted.
-      */
+      /**
+       * Deactivate the FLUTE session
+       *
+       * If the Transmitter is currently active then the FLUTE stream is halted and the state is changed to deactivated. Sending of
+       * packets will be halted until the activate() method is called. Note that this will pause File transmission part way through
+       * if a File is currently being transmitted.
+       */
       void deactivate();
 
+      /**
+       * Deactivate the FLUTE session
+       *
+       * When @a finish_file_transmissions is `true` the Transmitter will remain active until the queued transmissions have
+       * completed and will then become inactive. This allows applications to request deactivation without waiting for completion
+       * callbacks and checking number_of_files().
+       *
+       * @param finish_file_transmissions If `true`, defer deactivation until all queued transmissions complete. If `false`, halt
+       *        transmission immediately.
+       */
+      void deactivate(bool finish_file_transmissions);
+
      /**
       * Get number of files currently in queue for sending
       *
@@ -640,6 +651,8 @@ namespace LibFlute {
       void fdt_send_tick(const boost::system::error_code& error);
       void start_fdt_repeat_timer();
 
+      void _complete_deactivation();
+
       void file_transmitted(uint32_t toi);
 
       void handle_send_to(const boost::system::error_code& error);
@@ -670,7 +683,8 @@ namespace LibFlute {
       std::optional<boost::asio::ip::udp::endpoint> _tunnel_endpoint = std::nullopt;
       boost::asio::ip::address _tunnel_local_address;
 
-      bool _active;
+      std::atomic<bool> _active;
+      std::atomic<bool> _deactivate_when_all_files_sent = false;
   };
 
 } // end namespace LibFlute

src/Transmitter.cpp

@@ -722,6 +722,13 @@ auto Transmitter::file_transmitted(uint32_t toi) -> void
       _completion_cb(toi);
     }
   }
+
+  {
+    std::lock_guard<std::mutex> guard(_files_mutex);
+    if (_deactivate_when_all_files_sent && _files.empty()) {
+      _complete_deactivation();
+    }
+  }
 }
 
 auto Transmitter::send_next_packet() -> void
@@ -807,21 +814,44 @@ auto Transmitter::send_next_packet() -> void
 auto Transmitter::activate() -> void
 {
   if (!_active) {
+    _deactivate_when_all_files_sent = false;
     _active = true;
     start_fdt_repeat_timer();
     send_next_packet();
   }
 }
 
 auto Transmitter::deactivate() -> void
+{
+  deactivate(false);
+}
+
+auto Transmitter::deactivate(bool finish_file_transmissions) -> void
 {
   if (_active) {
-    _active = false;
-    _fdt_timer.cancel();
-    _send_timer.cancel();
+    if (finish_file_transmissions) {
+      std::lock_guard<std::mutex> guard(_files_mutex);
+      if (!_files.empty()) {
+        _deactivate_when_all_files_sent = true;
+        return;
+      }
+
+      _complete_deactivation();
+      return;
+    }
+
+    _complete_deactivation();
   }
 }
 
+auto Transmitter::_complete_deactivation() -> void
+{
+  _deactivate_when_all_files_sent = false;
+  _active = false;
+  _fdt_timer.cancel();
+  _send_timer.cancel();
+}
+
 auto Transmitter::start_fdt_repeat_timer() -> void
 {
     _fdt_timer.expires_from_now(boost::posix_time::seconds(_fdt_repeat_interval));

tests/test_transmitter.cpp

@@ -1,5 +1,12 @@
 #include <gtest/gtest.h>
+
 #include <boost/asio.hpp>
+
+#include <chrono>
+#include <future>
+#include <thread>
+#include <vector>
+
 #include "Transmitter.h"
 
 using namespace LibFlute;
@@ -73,3 +80,55 @@ TEST(TransmitterGetterSetterTest, UdpTunnelAddressSetAndUnset) {
   tx->udp_tunnel_address(std::nullopt);
   EXPECT_FALSE(tx->udp_tunnel_address().has_value());
 }
+
+TEST(TransmitterLifecycleTest, DeferredDeactivationDrainsQueuedFilesAndStopsFutureSends) {
+  using namespace std::chrono_literals;
+
+  boost::asio::io_context io;
+  auto work_guard = boost::asio::make_work_guard(io);
+  Transmitter tx("127.0.0.1", 5000, /*tsi*/1234, /*mtu*/1400, /*rate_limit*/0, io);
+
+  std::promise<void> first_completion_promise;
+  std::promise<void> second_completion_promise;
+  std::promise<void> lifecycle_settled_promise;
+  auto first_completion = first_completion_promise.get_future();
+  auto second_completion = second_completion_promise.get_future();
+  auto lifecycle_settled = lifecycle_settled_promise.get_future();
+
+  tx.register_completion_callback(
+      [&](const uint32_t toi) {
+        if (toi == 1) {
+          first_completion_promise.set_value();
+          boost::asio::post(io, [&lifecycle_settled_promise]() {
+            lifecycle_settled_promise.set_value();
+          });
+        } else if (toi == 2) {
+          second_completion_promise.set_value();
+        }
+      });
+
+  const std::vector<char> first_payload{'f', 'i', 'r', 's', 't'};
+  const std::vector<char> second_payload{'s', 'e', 'c', 'o', 'n', 'd'};
+  const auto first_file = std::make_shared<Transmitter::FileDescription>(
+      "test/first.bin", first_payload);
+  const auto second_file = std::make_shared<Transmitter::FileDescription>(
+      "test/second.bin", second_payload);
+
+  EXPECT_EQ(tx.send(first_file), 1);
+  tx.deactivate(true);
+  tx.activate();
+
+  std::thread io_thread([&io]() { io.run(); });
+  EXPECT_EQ(first_completion.wait_for(2s), std::future_status::ready);
+  EXPECT_EQ(lifecycle_settled.wait_for(2s), std::future_status::ready);
+
+  EXPECT_EQ(tx.send(second_file), 2);
+  EXPECT_EQ(second_completion.wait_for(250ms), std::future_status::timeout);
+
+  tx.activate();
+  EXPECT_EQ(second_completion.wait_for(2s), std::future_status::ready);
+
+  work_guard.reset();
+  io.stop();
+  io_thread.join();
+}