Start of refactoring of H5 reader used in sf_replay

sf-replay/include/ReplayH5Reader.hpp

@@ -12,27 +12,30 @@ class ReplayH5Reader {
 
     const std::string device_;
     const std::string channel_name_;
+    const uint64_t stop_pulse_id_;
 
     H5::H5File current_file_;
     std::string current_filename_;
     H5::DataSet dset_metadata_;
     H5::DataSet dset_frame_;
 
-    std::unique_ptr<char[]> frame_buffer_ = std::make_unique<char[]>(
-        core_buffer::MODULE_N_BYTES * core_buffer::REPLAY_READ_BUFFER_SIZE);
-    std::unique_ptr<ModuleFrame[]> metadata_buffer_ =
-            std::make_unique<ModuleFrame[]>(core_buffer::FILE_MOD);
-
-    uint64_t buffer_start_pulse_id_ = 0;
-    uint64_t buffer_end_pulse_id_ = 0;
-    void prepare_buffer_for_pulse(const uint64_t pulse_id);
+    void load_buffers(const uint64_t pulse_id,
+                      const size_t n_pulses,
+                      ReplayModuleFrameBuffer* metadata,
+                      char* frame_buffer);
 
 public:
-    ReplayH5Reader(const std::string device, const std::string channel_name);
+    ReplayH5Reader(
+            const std::string device,
+            const std::string channel_name,
+            const uint64_t stop_pulse_id);
     virtual ~ReplayH5Reader();
+
     void close_file();
-    bool get_frame(
-            const uint64_t pulse_id, ModuleFrame* metadata, char* frame_buffer);
+    bool get_buffer(
+            const uint64_t pulse_id,
+            ReplayModuleFrameBuffer* metadata,
+            char* frame_buffer);
 };
 
 

sf-replay/src/ReplayH5Reader.cpp

@@ -9,7 +9,11 @@
 using namespace std;
 using namespace core_buffer;
 
-void ReplayH5Reader::prepare_buffer_for_pulse(const uint64_t pulse_id)
+void ReplayH5Reader::load_buffers(
+        const uint64_t pulse_id,
+        const size_t n_pulses,
+        ReplayModuleFrameBuffer* metadata,
+        char* frame_buffer)
 {
     auto pulse_filename = BufferUtils::get_filename(
             device_, channel_name_, pulse_id);
@@ -99,20 +103,25 @@ void ReplayH5Reader::close_file()
     }
 }
 
-bool ReplayH5Reader::get_frame(
-        const uint64_t pulse_id, ModuleFrame* metadata, char* frame_buffer)
+bool ReplayH5Reader::get_buffer(
+        const uint64_t pulse_id,
+        ReplayModuleFrameBuffer* metadata,
+        char* frame_buffer)
 {
-    prepare_buffer_for_pulse(pulse_id);
+    auto start_pulse_id = pulse_id;
+    auto n_pulses = REPLAY_READ_BUFFER_SIZE;
+    auto buffer_end_pulse_id = start_pulse_id + n_pulses - 1;
+
+    // The last read segment might not fill the complete buffer.
+    if (stop_pulse_id_ < buffer_end_pulse_id) {
+        // stop_pulse_id_ must be included in the stream.
+        buffer_end_pulse_id = stop_pulse_id_;
+        n_pulses = buffer_end_pulse_id - start_pulse_id + 1;
+    }
+
+    load_buffers(pulse_id, n_pulses, metadata, frame_buffer);
 
-    auto metadata_buffer_index = BufferUtils::get_file_frame_index(pulse_id);
-    memcpy(metadata,
-            &(metadata_buffer_[metadata_buffer_index]),
-            sizeof(ModuleFrame));
 
-    auto frame_buffer_index = pulse_id - buffer_start_pulse_id_;
-    memcpy(frame_buffer,
-            &(frame_buffer_[frame_buffer_index * MODULE_N_BYTES]),
-            MODULE_N_BYTES);
 
     if (metadata->pulse_id == 0) {
         // Signal that there is no frame at this pulse_id.
@@ -125,7 +134,7 @@ bool ReplayH5Reader::get_frame(
         using namespace date;
         using namespace chrono;
         err_msg << "[" << system_clock::now() << "]";
-        err_msg << "[ReplayH5Reader::get_frame]";
+        err_msg << "[ReplayH5Reader::get_buffer]";
         err_msg << " Corrupted file " << current_filename_;
         err_msg << " index_in_file " << metadata_buffer_index;
         err_msg << " expected pulse_id " << pulse_id;