MockAcquisitionDevice: More robust implementation

receiver/MockAcquisitionDevice.cpp

@@ -7,13 +7,15 @@
 
 void MockAcquisitionDevice::Start(const DiffractionExperiment& experiment) {
     idle = false;
+    cancel = false;
+    measure = std::async(std::launch::async, &MockAcquisitionDevice::MeasureThread, this);
 }
 
 void MockAcquisitionDevice::Cancel() {
     if (!idle) {
         if (logger)
             logger->Info("MockAcquisitionDevice cancelling " + std::to_string(data_stream));
-        Terminate();
+        cancel = true;
     }
 }
 
@@ -22,7 +24,6 @@ MockAcquisitionDevice::MockAcquisitionDevice(uint16_t data_stream, size_t in_fra
     max_modules = 16;
     MapBuffersStandard(in_frame_buffer_size_modules, 1, -1);
     max_handle = in_frame_buffer_size_modules;
-    work_completion_queue.Put(Completion{.type = Completion::Type::Start});
 }
 
 void MockAcquisitionDevice::SendCompletion(uint32_t handle, uint16_t module_number, uint64_t frame_number) {
@@ -34,7 +35,7 @@ void MockAcquisitionDevice::SendCompletion(uint32_t handle, uint16_t module_numb
     c.packet_mask[0] = UINT64_MAX;
     c.packet_mask[1] = UINT64_MAX;
     c.packet_count = 128;
-    work_completion_queue.Put(c);
+    mock_completions.Put(c);
 }
 
 void MockAcquisitionDevice::AddModule(uint64_t frame_number, uint16_t module_number, const uint16_t *data) {
@@ -52,13 +53,33 @@ void MockAcquisitionDevice::AddModule(uint64_t frame_number, uint16_t module_num
     }
 }
 
-void MockAcquisitionDevice::Terminate() {
+void MockAcquisitionDevice::MeasureThread() {
+    work_completion_queue.Put(Completion{.type = Completion::Type::Start});
+
+    while (!cancel) {
+        Completion c{};
+
+        if (mock_completions.Get(c)) {
+            if (c.type == Completion::Type::Image)
+                work_completion_queue.Put(c);
+            else
+                cancel = true;
+        } else
+            std::this_thread::sleep_for(std::chrono::microseconds(100));
+    }
+
     work_completion_queue.Put(Completion{
             .type = Completion::Type::End
     });
     idle = true;
 }
 
+void MockAcquisitionDevice::Terminate() {
+    mock_completions.Put(Completion{
+            .type = Completion::Type::End
+    });
+}
+
 std::string MockAcquisitionDevice::GetMACAddress() const {
     return "00:00:00:00:00:00";
 }
@@ -67,3 +88,7 @@ std::string MockAcquisitionDevice::GetIPv4Address() const {
     return "127.0.0.1";
 }
 
+void MockAcquisitionDevice::Finalize() {
+    if (measure.valid())
+        measure.get();
+}

receiver/MockAcquisitionDevice.h

@@ -6,15 +6,21 @@
 
 #include "AcquisitionDevice.h"
 #include "../common/ThreadSafeFIFO.h"
+#include <future>
 
 class MockAcquisitionDevice : public AcquisitionDevice {
     uint32_t current_handle = 0;
     uint32_t max_handle = 0;
 
     bool idle = true;
+    std::future<void> measure;
+    volatile bool cancel = false;
+
+    ThreadSafeFIFO<Completion> mock_completions;
 
     void SendCompletion(uint32_t handle, uint16_t module_number, uint64_t frame_number);
     void Start(const DiffractionExperiment& experiment) override;
+    void MeasureThread();
 public:
     MockAcquisitionDevice(uint16_t data_stream, size_t in_frame_buffer_size_modules);
     void AddModule(uint64_t frame, uint16_t module_number, const uint16_t *data);
@@ -22,6 +28,7 @@ public:
     std::string GetMACAddress() const override;
     std::string GetIPv4Address() const override;
     void Cancel() override;
+    void Finalize() override;
 };
 
 #endif //JUNGFRAUJOCH_MOCKACQUISITIONDEVICE_H