Fix filter tests

src/tests/unit-tests/signal-processing-blocks/filter/notch_filter_lite_test.cc

@@ -23,6 +23,7 @@
 #include <chrono>
 #include <complex>
 #include <cstdint>
+#include <thread>
 #ifdef GR_GREATER_38
 #include <gnuradio/analog/sig_source.h>
 #else
@@ -54,6 +55,14 @@ protected:
     }
     ~NotchFilterLiteTest() override = default;
 
+    bool stop = false;
+    std::thread ch_thread;
+    void start_queue();
+    void wait_message();
+    void process_message();
+    void stop_queue();
+    pmt::pmt_t message;
+
     void init();
     void configure_gr_complex_gr_complex();
     std::shared_ptr<Concurrent_Queue<pmt::pmt_t>> queue;
@@ -64,6 +73,36 @@ protected:
 };
 
 
+void NotchFilterLiteTest::start_queue()
+{
+    stop = false;
+    ch_thread = std::thread(&NotchFilterLiteTest::wait_message, this);
+}
+
+
+void NotchFilterLiteTest::wait_message()
+{
+    while (!stop)
+        {
+            queue->wait_and_pop(message);
+            process_message();
+        }
+}
+
+
+void NotchFilterLiteTest::process_message()
+{
+    stop_queue();
+    top_block->stop();
+}
+
+
+void NotchFilterLiteTest::stop_queue()
+{
+    stop = true;
+}
+
+
 void NotchFilterLiteTest::init()
 {
     config->set_property("InputFilter.pfa", "0.01");
@@ -117,12 +156,14 @@ TEST_F(NotchFilterLiteTest, ConnectAndRun)
         top_block->connect(filter->get_right_block(), 0, null_sink, 0);
     }) << "Failure connecting the top_block.";
 
+    start_queue();
     EXPECT_NO_THROW({
         start = std::chrono::system_clock::now();
         top_block->run();  // Start threads and wait
         end = std::chrono::system_clock::now();
         elapsed_seconds = end - start;
     }) << "Failure running the top_block.";
+    ch_thread.join();
     std::cout << "Filtered " << nsamples << " samples in " << elapsed_seconds.count() * 1e6 << " microseconds\n";
 }
 
@@ -159,11 +200,13 @@ TEST_F(NotchFilterLiteTest, ConnectAndRunGrcomplex)
         top_block->connect(filter->get_right_block(), 0, null_sink, 0);
     }) << "Failure connecting the top_block.";
 
+    start_queue();
     EXPECT_NO_THROW({
         start = std::chrono::system_clock::now();
         top_block->run();  // Start threads and wait
         end = std::chrono::system_clock::now();
         elapsed_seconds = end - start;
     }) << "Failure running the top_block.";
+    ch_thread.join();
     std::cout << "Filtered " << nsamples << " gr_complex samples in " << elapsed_seconds.count() * 1e6 << " microseconds\n";
 }

src/tests/unit-tests/signal-processing-blocks/filter/notch_filter_test.cc

@@ -23,6 +23,7 @@
 #include <chrono>
 #include <complex>
 #include <cstdint>
+#include <thread>
 #ifdef GR_GREATER_38
 #include <gnuradio/analog/sig_source.h>
 #else
@@ -54,6 +55,14 @@ protected:
     }
     ~NotchFilterTest() override = default;
 
+    bool stop = false;
+    std::thread ch_thread;
+    void start_queue();
+    void wait_message();
+    void process_message();
+    void stop_queue();
+    pmt::pmt_t message;
+
     void init();
     void configure_gr_complex_gr_complex();
     std::shared_ptr<Concurrent_Queue<pmt::pmt_t>> queue;
@@ -64,6 +73,36 @@ protected:
 };
 
 
+void NotchFilterTest::start_queue()
+{
+    stop = false;
+    ch_thread = std::thread(&NotchFilterTest::wait_message, this);
+}
+
+
+void NotchFilterTest::wait_message()
+{
+    while (!stop)
+        {
+            queue->wait_and_pop(message);
+            process_message();
+        }
+}
+
+
+void NotchFilterTest::process_message()
+{
+    stop_queue();
+    top_block->stop();
+}
+
+
+void NotchFilterTest::stop_queue()
+{
+    stop = true;
+}
+
+
 void NotchFilterTest::init()
 {
     config->set_property("InputFilter.pfa", "0.01");
@@ -116,13 +155,14 @@ TEST_F(NotchFilterTest, ConnectAndRun)
         top_block->connect(valve, 0, filter->get_left_block(), 0);
         top_block->connect(filter->get_right_block(), 0, null_sink, 0);
     }) << "Failure connecting the top_block.";
-
+    start_queue();
     EXPECT_NO_THROW({
         start = std::chrono::system_clock::now();
         top_block->run();  // Start threads and wait
         end = std::chrono::system_clock::now();
         elapsed_seconds = end - start;
     }) << "Failure running the top_block.";
+    ch_thread.join();
     std::cout << "Filtered " << nsamples << " samples in " << elapsed_seconds.count() * 1e6 << " microseconds\n";
 }
 
@@ -158,12 +198,13 @@ TEST_F(NotchFilterTest, ConnectAndRunGrcomplex)
         top_block->connect(source->get_right_block(), 0, filter->get_left_block(), 0);
         top_block->connect(filter->get_right_block(), 0, null_sink, 0);
     }) << "Failure connecting the top_block.";
-
+    start_queue();
     EXPECT_NO_THROW({
         start = std::chrono::system_clock::now();
         top_block->run();  // Start threads and wait
         end = std::chrono::system_clock::now();
         elapsed_seconds = end - start;
     }) << "Failure running the top_block.";
+    ch_thread.join();
     std::cout << "Filtered " << nsamples << " gr_complex samples in " << elapsed_seconds.count() * 1e6 << " microseconds\n";
 }

src/tests/unit-tests/signal-processing-blocks/filter/pulse_blanking_filter_test.cc

@@ -23,6 +23,7 @@
 #include <chrono>
 #include <complex>
 #include <cstdint>
+#include <thread>
 #ifdef GR_GREATER_38
 #include <gnuradio/analog/sig_source.h>
 #else
@@ -53,7 +54,12 @@ protected:
         nsamples = FLAGS_pb_filter_test_nsamples;
     }
     ~PulseBlankingFilterTest() override = default;
-
+    bool stop = false;
+    std::thread ch_thread;
+    void start_queue();
+    void wait_message();
+    void process_message();
+    void stop_queue();
     void init();
     void configure_gr_complex_gr_complex();
     std::shared_ptr<Concurrent_Queue<pmt::pmt_t>> queue;
@@ -61,9 +67,40 @@ protected:
     std::shared_ptr<InMemoryConfiguration> config;
     size_t item_size;
     int nsamples;
+    pmt::pmt_t message;
 };
 
 
+void PulseBlankingFilterTest::start_queue()
+{
+    stop = false;
+    ch_thread = std::thread(&PulseBlankingFilterTest::wait_message, this);
+}
+
+
+void PulseBlankingFilterTest::wait_message()
+{
+    while (!stop)
+        {
+            queue->wait_and_pop(message);
+            process_message();
+        }
+}
+
+
+void PulseBlankingFilterTest::process_message()
+{
+    stop_queue();
+    top_block->stop();
+}
+
+
+void PulseBlankingFilterTest::stop_queue()
+{
+    stop = true;
+}
+
+
 void PulseBlankingFilterTest::init()
 {
     config->set_property("InputFilter.pfa", "0.04");
@@ -115,13 +152,14 @@ TEST_F(PulseBlankingFilterTest, ConnectAndRun)
         top_block->connect(valve, 0, filter->get_left_block(), 0);
         top_block->connect(filter->get_right_block(), 0, null_sink, 0);
     }) << "Failure connecting the top_block.";
-
+    start_queue();
     EXPECT_NO_THROW({
         start = std::chrono::system_clock::now();
         top_block->run();  // Start threads and wait
         end = std::chrono::system_clock::now();
         elapsed_seconds = end - start;
     }) << "Failure running the top_block.";
+    ch_thread.join();
     std::cout << "Filtered " << nsamples << " samples in " << elapsed_seconds.count() * 1e6 << " microseconds\n";
 }
 
@@ -157,12 +195,13 @@ TEST_F(PulseBlankingFilterTest, ConnectAndRunGrcomplex)
         top_block->connect(source->get_right_block(), 0, filter->get_left_block(), 0);
         top_block->connect(filter->get_right_block(), 0, null_sink, 0);
     }) << "Failure connecting the top_block.";
-
+    start_queue();
     EXPECT_NO_THROW({
         start = std::chrono::system_clock::now();
         top_block->run();  // Start threads and wait
         end = std::chrono::system_clock::now();
         elapsed_seconds = end - start;
     }) << "Failure running the top_block.";
+    ch_thread.join();
     std::cout << "Filtered " << nsamples << " gr_complex samples in " << elapsed_seconds.count() * 1e6 << " microseconds\n";
 }