diff --git a/.gitignore b/.gitignore
index 098f81eba37b3421a0af9d3fae77724418fd5433..9a7e609c29e483635f722b3206c6322d3f042efc 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,7 +10,8 @@ pipeline.gdsl
*.j2~
*.h~
*.cxx~
-/integration-tests/output-files/*.nxs
+/integration-tests/output-files/*.*
+/domain-tests/output_files/*.*
venv
*.log
*.DS_Store
diff --git a/domain-tests/data_files/buffered_data.json b/domain-tests/data_files/buffered_data.json
new file mode 100644
index 0000000000000000000000000000000000000000..919dbb570eda918e4e96f716223d6532d74a031b
--- /dev/null
+++ b/domain-tests/data_files/buffered_data.json
@@ -0,0 +1,18 @@
+[
+ {
+ "schema": "f144",
+ "topic": "local_choppers",
+ "kafka_timestamp": 5000,
+ "source_name": "local:choppers:delay",
+ "timestamp": 5000,
+ "value": 2
+ },
+ {
+ "schema": "f144",
+ "topic": "local_choppers",
+ "kafka_timestamp": 9000,
+ "source_name": "local:choppers:delay",
+ "timestamp": 9000,
+ "value": 3
+ }
+]
diff --git a/domain-tests/data_files/messages_before_and_after_data.json b/domain-tests/data_files/messages_before_and_after_data.json
index da4e87128b3062f59047a784705659e7a0ed3a78..7140c5078c879a9c4e1d19b4cc27efdb07202893 100644
--- a/domain-tests/data_files/messages_before_and_after_data.json
+++ b/domain-tests/data_files/messages_before_and_after_data.json
@@ -102,31 +102,41 @@
{
"schema": "ev44",
"topic": "local_detector",
- "kafka_timestamp": 11000,
+ "kafka_timestamp": 10000,
"source_name": "detector_events",
"message_id": 667,
- "reference_time": 11000,
+ "reference_time": 10000,
"time_of_flight": [50, 60, 70, 80],
"pixel_ids": [1, 2, 3, 4]
},
{
"schema": "ev44",
"topic": "local_detector",
- "kafka_timestamp": 16000,
+ "kafka_timestamp": 12000,
"source_name": "detector_events",
"message_id": 668,
- "reference_time": 16000,
+ "reference_time": 12000,
"time_of_flight": [90, 100, 110, 120],
"pixel_ids": [1, 2, 3, 4]
},
{
"schema": "ev44",
"topic": "local_detector",
- "kafka_timestamp": 17000,
+ "kafka_timestamp": 15000,
"source_name": "detector_events",
"message_id": 669,
- "reference_time": 17000,
+ "reference_time": 15000,
"time_of_flight": [130, 140, 150, 160],
"pixel_ids": [1, 2, 3, 4]
+ },
+ {
+ "schema": "ev44",
+ "topic": "local_detector",
+ "kafka_timestamp": 16000,
+ "source_name": "detector_events",
+ "message_id": 670,
+ "reference_time": 16000,
+ "time_of_flight": [170, 180, 190, 200],
+ "pixel_ids": [1, 2, 3, 4]
}
]
diff --git a/domain-tests/nexus_templates/buffered_template.json b/domain-tests/nexus_templates/buffered_template.json
new file mode 100644
index 0000000000000000000000000000000000000000..b6ec0427f91acde6d7d0cd46079b70f4de9f7ab4
--- /dev/null
+++ b/domain-tests/nexus_templates/buffered_template.json
@@ -0,0 +1,89 @@
+{
+ "children": [
+ {
+ "name": "entry",
+ "type": "group",
+ "attributes": [
+ {
+ "name": "NX_class",
+ "dtype": "string",
+ "values": "NXentry"
+ }
+ ],
+ "children": [
+ {
+ "module": "dataset",
+ "config": {
+ "name": "title",
+ "values": "This is a title",
+ "dtype": "string"
+ }
+ },
+ {
+ "module": "mdat",
+ "config": {
+ "items": [
+ "start_time",
+ "end_time"
+ ]
+ }
+ },
+ {
+ "name": "instrument",
+ "type": "group",
+ "attributes": [
+ {
+ "name": "NX_class",
+ "dtype": "string",
+ "values": "NXinstrument"
+ }
+ ],
+ "children": [
+ {
+ "name": "chopper",
+ "type": "group",
+ "attributes": [
+ {
+ "name": "NX_class",
+ "dtype": "string",
+ "values": "NXdisk_chopper"
+ }
+ ],
+ "children": [
+ {
+ "name": "delay",
+ "type": "group",
+ "attributes": [
+ {
+ "name": "NX_class",
+ "dtype": "string",
+ "values": "NXlog"
+ }
+ ],
+ "children": [
+ {
+ "module": "f144",
+ "config": {
+ "source": "local:choppers:delay",
+ "topic": "local_choppers",
+ "dtype": "double"
+ }
+ }
+ ]
+ },
+ {
+ "module": "dataset",
+ "config": {
+ "name": "depends_on",
+ "values": ".",
+ "dtype": "string"
+ }
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+}
diff --git a/domain-tests/test_buffered.py b/domain-tests/test_buffered.py
new file mode 100644
index 0000000000000000000000000000000000000000..1c5301893dcee8000302ad157d1977ef62814be2
--- /dev/null
+++ b/domain-tests/test_buffered.py
@@ -0,0 +1,22 @@
+import h5py
+import pytest
+from conftest import write_file
+
+
+# This fixture is used to create the file
+@pytest.fixture(scope="module")
+def local_file(request):
+ return write_file(
+ request,
+ "output_files/buffered.hdf",
+ "nexus_templates/buffered_template.json",
+ "data_files/buffered_data.json",
+ )
+
+
+def test_f144_data_before_start_is_written_when_no_data_between_start_and_stop(
+ local_file,
+):
+ with h5py.File(local_file, "r") as f:
+ assert f["/entry/instrument/chopper/delay/value"][0] == 3
+ assert f["/entry/instrument/chopper/delay/time"][0] == 9_000_000_000
diff --git a/domain-tests/test_messages_before_and_after.py b/domain-tests/test_messages_before_and_after.py
index 4a858cb667e5f07279b82f161670371f0cb43921..cf3c0dd022c10c5da643339e7a24c521e8151197 100644
--- a/domain-tests/test_messages_before_and_after.py
+++ b/domain-tests/test_messages_before_and_after.py
@@ -1,5 +1,4 @@
import h5py
-import numpy as np
import pytest
from conftest import write_file
@@ -21,23 +20,23 @@ def test_last_f144_data_before_start_is_written_but_earlier_values_ignored(local
assert f["/entry/instrument/chopper/delay/time"][0] == 9_999_000_000
-def test_first_f144_data_after_stop_written_but_later_values_ignored(local_file):
+def test_f144_data_after_stop_not_written(local_file):
with h5py.File(local_file, "r") as f:
- assert f["/entry/instrument/chopper/delay/value"][~0] == 17
- assert f["/entry/instrument/chopper/delay/time"][~0] == 16_000_000_000
+ assert f["/entry/instrument/chopper/delay/value"][~0] == 16
+ assert f["/entry/instrument/chopper/delay/time"][~0] == 15_000_000_000
-def test_last_ev44_data_before_start_is_written_but_earlier_values_ignored(local_file):
+def test_ev44_data_before_start_is_not_written(local_file):
with h5py.File(local_file, "r") as f:
assert (
f["/entry/instrument/event_detector/events/event_time_zero"][0]
- == 9_000_000_000
+ == 10_000_000_000
)
-def test_first_ev44_data_after_stop_written_but_later_values_ignored(local_file):
+def test_ev44_data_after_stop_not_written(local_file):
with h5py.File(local_file, "r") as f:
assert (
f["/entry/instrument/event_detector/events/event_time_zero"][~0]
- == 16_000_000_000
+ == 15_000_000_000
)
diff --git a/integration-tests/test_filewriter.py b/integration-tests/test_filewriter.py
index 5633fc25a652f6b8fd34e8dead0df053b605c58e..211d041d64da5ebb4544d0d4c589d9feb6dcb72e 100644
--- a/integration-tests/test_filewriter.py
+++ b/integration-tests/test_filewriter.py
@@ -341,9 +341,10 @@ class TestFileWriter:
def test_data_before_and_after(self, file_writer):
"""
Test that when writing in the past:
- - the last value before the start time is written
- - the first value after the stop time is written
- - all values inbetween are written
+ - the last value before the start time is written for f144
+ - the first value after the stop time is written for f144
+ - all f144 values inbetween are written
+ - ev44 only writes the values inbetween the start and stop times
"""
producer = Producer({"bootstrap.servers": ",".join(get_brokers())})
@@ -379,12 +380,12 @@ class TestFileWriter:
with h5py.File(os.path.join(OUTPUT_DIR, file_name), "r") as f:
# Check data is as expected
- assert len(f["/entry/detector/event_time_zero"]) == 8
- assert f["/entry/detector/event_id"][0] == 15
- assert f["/entry/detector/event_id"][~0] == 54
- assert len(f["/entry/motion/time"]) == 8
+ assert len(f["/entry/detector/event_time_zero"]) == 6
+ assert f["/entry/detector/event_id"][0] == 20
+ assert f["/entry/detector/event_id"][~0] == 49
+ assert len(f["/entry/motion/time"]) == 7
assert f["/entry/motion/value"][0] == 3
- assert f["/entry/motion/value"][~0] == 10
+ assert f["/entry/motion/value"][~0] == 9
def test_start_and_stop_in_the_future(self, file_writer):
"""
diff --git a/src/Stream/MessageWriter.cpp b/src/Stream/MessageWriter.cpp
index 2625bf879b75eb71a7aeb075db41438cedbb6b31..f02e36a435285064d2a26dd6268360f0650b03fc 100644
--- a/src/Stream/MessageWriter.cpp
+++ b/src/Stream/MessageWriter.cpp
@@ -51,16 +51,18 @@ MessageWriter::~MessageWriter() {
}
}
-void MessageWriter::addMessage(Message const &Msg) {
- WriteJobs.enqueue([=]() { writeMsgImpl(Msg.DestPtr, Msg.FbMsg); });
+void MessageWriter::addMessage(Message const &Msg, bool is_buffered_message) {
+ WriteJobs.enqueue(
+ [=]() { writeMsgImpl(Msg.DestPtr, Msg.FbMsg, is_buffered_message); });
}
void MessageWriter::stop() { RunThread.store(false); }
void MessageWriter::writeMsgImpl(WriterModule::Base *ModulePtr,
- FileWriter::FlatbufferMessage const &Msg) {
+ FileWriter::FlatbufferMessage const &Msg,
+ bool is_buffered_message) {
try {
- ModulePtr->write(Msg);
+ ModulePtr->write(Msg, is_buffered_message);
WritesDone++;
} catch (WriterModule::WriterException &E) {
WriteErrors++;
diff --git a/src/Stream/MessageWriter.h b/src/Stream/MessageWriter.h
index 9ef768597f0f20e36a49f8a642d1092482d11d2b..e7e75687c7ab04bec64be496806ae2d8d2dd78a4 100644
--- a/src/Stream/MessageWriter.h
+++ b/src/Stream/MessageWriter.h
@@ -38,7 +38,7 @@ public:
virtual ~MessageWriter();
- virtual void addMessage(Message const &Msg);
+ virtual void addMessage(Message const &Msg, bool is_buffered_message);
/// \brief Tell the writer thread to stop.
///
@@ -60,7 +60,8 @@ public:
protected:
virtual void writeMsgImpl(WriterModule::Base *ModulePtr,
- FileWriter::FlatbufferMessage const &Msg);
+ FileWriter::FlatbufferMessage const &Msg,
+ bool is_buffered_message);
virtual void threadFunction();
virtual void flushData() { FlushDataFunction(); };
diff --git a/src/Stream/SourceFilter.cpp b/src/Stream/SourceFilter.cpp
index 77a5cd282a4f40992a5309bedfd3f9c0da6822bd..586e54426fb475299b5d99ff98ab59f936148c01 100644
--- a/src/Stream/SourceFilter.cpp
+++ b/src/Stream/SourceFilter.cpp
@@ -36,7 +36,7 @@ bool SourceFilter::has_finished() const { return _is_finished; }
void SourceFilter::forward_buffered_message() {
if (_buffered_message.isValid()) {
- forward_message(_buffered_message);
+ forward_message(_buffered_message, true);
_buffered_message = FileWriter::FlatbufferMessage();
}
}
@@ -86,17 +86,19 @@ bool SourceFilter::filter_message(
}
if (message_time > _stop_time) {
_is_finished = true;
+ forward_buffered_message();
+ return false;
}
forward_buffered_message();
forward_message(message);
return true;
}
-void SourceFilter::forward_message(
- FileWriter::FlatbufferMessage const &message) {
+void SourceFilter::forward_message(FileWriter::FlatbufferMessage const &message,
+ bool is_buffered_message) {
++MessagesTransmitted;
for (auto const &writer_module : _destination_writer_modules) {
- _writer->addMessage({writer_module, message});
+ _writer->addMessage({writer_module, message}, is_buffered_message);
}
}
diff --git a/src/Stream/SourceFilter.h b/src/Stream/SourceFilter.h
index 60b63f61a26b494b6dce41e8893a74837042a35c..443ce20dac28b387e19122af93fc4920d42bd936 100644
--- a/src/Stream/SourceFilter.h
+++ b/src/Stream/SourceFilter.h
@@ -58,7 +58,8 @@ public:
}
private:
- void forward_message(FileWriter::FlatbufferMessage const &message);
+ void forward_message(FileWriter::FlatbufferMessage const &message,
+ bool is_buffered_message = false);
void forward_buffered_message();
time_point _start_time;
time_point _stop_time;
diff --git a/src/WriterModule/ad00/ad00_Writer.cpp b/src/WriterModule/ad00/ad00_Writer.cpp
index 10524503b601f55616a1b3cdf9b53edd5fd46e73..b50951a643975df8854ceb7c10da70becb9fe47e 100644
--- a/src/WriterModule/ad00/ad00_Writer.cpp
+++ b/src/WriterModule/ad00/ad00_Writer.cpp
@@ -139,7 +139,8 @@ void msgTypeIsConfigType(ad00_Writer::Type ConfigType, DType MsgType) {
}
}
-void ad00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
+bool ad00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message,
+ [[maybe_unused]] bool is_buffered_message) {
auto ad00 = Getad00_ADArray(Message.data());
auto DataShape =
hdf5::Dimensions(ad00->dimensions()->begin(), ad00->dimensions()->end());
@@ -199,6 +200,7 @@ void ad00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
CueTimestamp.appendElement(CurrentTimestamp);
CueCounter = 0;
}
+ return true;
}
template <typename Type>
diff --git a/src/WriterModule/ad00/ad00_Writer.h b/src/WriterModule/ad00/ad00_Writer.h
index 1975a35b13043f6384974ef180314d8372f41210..5922457d3cbd68cca1591ab33797c4197597e783 100644
--- a/src/WriterModule/ad00/ad00_Writer.h
+++ b/src/WriterModule/ad00/ad00_Writer.h
@@ -33,7 +33,8 @@ public:
InitResult reopen(hdf5::node::Group &HDFGroup) override;
- void writeImpl(FileWriter::FlatbufferMessage const &Message) override;
+ bool writeImpl(FileWriter::FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
enum class Type {
int8,
diff --git a/src/WriterModule/al00/al00_Writer.cpp b/src/WriterModule/al00/al00_Writer.cpp
index ac56b12ac720f3dea9c73478b3e2d7776e789a3c..9beeff2baa374265bcba4601256a7c638dbea470 100644
--- a/src/WriterModule/al00/al00_Writer.cpp
+++ b/src/WriterModule/al00/al00_Writer.cpp
@@ -54,7 +54,8 @@ InitResult al00_Writer::reopen(hdf5::node::Group &HDFGroup) {
return InitResult::OK;
}
-void al00_Writer::writeImpl(FlatbufferMessage const &Message) {
+bool al00_Writer::writeImpl(FlatbufferMessage const &Message,
+ [[maybe_unused]] bool is_buffered_message) {
auto AlarmMessage = GetAlarm(Message.data());
AlarmTime.appendElement(AlarmMessage->timestamp());
@@ -65,6 +66,7 @@ void al00_Writer::writeImpl(FlatbufferMessage const &Message) {
AlarmMessageString = "NO ALARM MESSAGE";
}
AlarmMsg.appendStringElement(AlarmMessageString);
+ return true;
}
/// Register the writer module.
diff --git a/src/WriterModule/al00/al00_Writer.h b/src/WriterModule/al00/al00_Writer.h
index 87ce75656ccd639d1173ec0c85aa3bc86be8aeef..fa68a2b36f74de16e57de9016e801515dd20832a 100644
--- a/src/WriterModule/al00/al00_Writer.h
+++ b/src/WriterModule/al00/al00_Writer.h
@@ -33,7 +33,8 @@ public:
WriterModule::InitResult reopen(hdf5::node::Group &HDFGroup) override;
/// Write an incoming message which should contain a flatbuffer.
- void writeImpl(FlatbufferMessage const &Message) override;
+ bool writeImpl(FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
al00_Writer() : WriterModule::Base("al00", false, "NXlog") {}
~al00_Writer() override = default;
diff --git a/src/WriterModule/da00/da00_Writer.cpp b/src/WriterModule/da00/da00_Writer.cpp
index eb94cb8a5d48795c705977e5b7574f789ed4b40a..c36674bfe3b2d8182dab8d6d9c353f36a50d7fb0 100644
--- a/src/WriterModule/da00/da00_Writer.cpp
+++ b/src/WriterModule/da00/da00_Writer.cpp
@@ -330,7 +330,8 @@ InitResult da00_Writer::reopen(hdf5::node::Group &HDFGroup) {
return InitResult::OK;
}
-void da00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
+bool da00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message,
+ [[maybe_unused]] bool is_buffered_message) {
const auto da00_obj = Getda00_DataArray(Message.data());
if (isFirstMessage) {
handle_first_message(da00_obj);
@@ -382,6 +383,7 @@ void da00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
CueTimestampZero.appendElement(da00_obj->timestamp());
CueCounter = 0;
}
+ return true;
}
} // namespace WriterModule::da00
diff --git a/src/WriterModule/da00/da00_Writer.h b/src/WriterModule/da00/da00_Writer.h
index c07d987bb7657b56dc74f5f5820f8e04baeb50f8..eb5301e557aa9787f55735bebe7b147fcb3df035 100644
--- a/src/WriterModule/da00/da00_Writer.h
+++ b/src/WriterModule/da00/da00_Writer.h
@@ -36,7 +36,8 @@ public:
InitResult reopen(hdf5::node::Group &HDFGroup) override;
- void writeImpl(FileWriter::FlatbufferMessage const &Message) override;
+ bool writeImpl(FileWriter::FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
NeXusDataset::Time Timestamp;
NeXusDataset::CueIndex CueIndex;
diff --git a/src/WriterModule/ep01/ep01_Writer.cpp b/src/WriterModule/ep01/ep01_Writer.cpp
index 6b13d578603497a91966ff9ae6b36cc9075a80af..3e6dc1f313c1bb0c236ca86500875e00f0e2c312 100644
--- a/src/WriterModule/ep01/ep01_Writer.cpp
+++ b/src/WriterModule/ep01/ep01_Writer.cpp
@@ -35,12 +35,14 @@ InitResult ep01_Writer::init_hdf(hdf5::node::Group &HDFGroup) {
return InitResult::OK;
}
-void ep01_Writer::writeImpl(FileWriter::FlatbufferMessage const &Message) {
+bool ep01_Writer::writeImpl(FileWriter::FlatbufferMessage const &Message,
+ [[maybe_unused]] bool is_buffered_message) {
auto FlatBuffer = GetEpicsPVConnectionInfo(Message.data());
std::int16_t Status = static_cast<std::int16_t>(FlatBuffer->status());
StatusDataset.appendElement(Status);
auto FBTimestamp = FlatBuffer->timestamp();
TimestampDataset.appendElement(FBTimestamp);
+ return true;
}
static WriterModule::Registry::Registrar<ep01_Writer>
diff --git a/src/WriterModule/ep01/ep01_Writer.h b/src/WriterModule/ep01/ep01_Writer.h
index de0537c92ba23d45fd96af06830b23acd904a2c9..b13be6d9d5c6d76153a93c442103100026e5a939 100644
--- a/src/WriterModule/ep01/ep01_Writer.h
+++ b/src/WriterModule/ep01/ep01_Writer.h
@@ -9,7 +9,8 @@ class ep01_Writer final : public WriterModule::Base {
public:
InitResult init_hdf(hdf5::node::Group &HDFGroup) override;
InitResult reopen(hdf5::node::Group &HDFGroup) override;
- void writeImpl(FileWriter::FlatbufferMessage const &Message) override;
+ bool writeImpl(FileWriter::FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
ep01_Writer() : WriterModule::Base("ep01", false, "NXlog") {}
~ep01_Writer() override = default;
diff --git a/src/WriterModule/ev44/ev44_Writer.cpp b/src/WriterModule/ev44/ev44_Writer.cpp
index b7a30884693b1bcb1047e2c11b37f23f91abb4cc..b47611f2e59a485c17d31c615e30fa46df261664 100644
--- a/src/WriterModule/ev44/ev44_Writer.cpp
+++ b/src/WriterModule/ev44/ev44_Writer.cpp
@@ -89,7 +89,12 @@ WriterModule::InitResult ev44_Writer::reopen(hdf5::node::Group &HDFGroup) {
return WriterModule::InitResult::OK;
}
-void ev44_Writer::writeImpl(FlatbufferMessage const &Message) {
+bool ev44_Writer::writeImpl(FlatbufferMessage const &Message,
+ bool is_buffered_message) {
+ if (is_buffered_message) {
+ // Ignore buffered data for event data
+ return false;
+ }
auto EventMsgFlatbuffer = GetEvent44Message(Message.data());
auto CurrentNumberOfEvents = EventMsgFlatbuffer->time_of_flight()->size();
if (EventMsgFlatbuffer->pixel_id()->size() > 0 &&
@@ -129,6 +134,7 @@ void ev44_Writer::writeImpl(FlatbufferMessage const &Message) {
}
EventsWrittenMetadataField.setValue(EventsWritten);
}
+ return true;
}
void ev44_Writer::register_meta_data(const hdf5::node::Group &HDFGroup,
diff --git a/src/WriterModule/ev44/ev44_Writer.h b/src/WriterModule/ev44/ev44_Writer.h
index 0850818dc86b466f98a0a17e168c85a5acd77498..7e3d4649375e07f724330b1520cfac5fa56403be 100644
--- a/src/WriterModule/ev44/ev44_Writer.h
+++ b/src/WriterModule/ev44/ev44_Writer.h
@@ -27,7 +27,8 @@ public:
/// \brief Write flatbuffer message.
///
/// \param FlatBufferMessage
- void writeImpl(FlatbufferMessage const &Message) override;
+ bool writeImpl(FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
NeXusDataset::EventTimeOffset EventTimeOffset;
NeXusDataset::EventId EventId;
diff --git a/src/WriterModule/f144/f144_Writer.cpp b/src/WriterModule/f144/f144_Writer.cpp
index 7c21d83f0c5309fd9a032a877df6adadc7fc35cf..8707ac4d708fca3fb899b545ed7ea63af0f29755 100644
--- a/src/WriterModule/f144/f144_Writer.cpp
+++ b/src/WriterModule/f144/f144_Writer.cpp
@@ -194,7 +194,8 @@ void msgTypeIsConfigType(f144_Writer::Type ConfigType, Value MsgType) {
}
}
-void f144_Writer::writeImpl(FlatbufferMessage const &Message) {
+bool f144_Writer::writeImpl(FlatbufferMessage const &Message,
+ [[maybe_unused]] bool is_buffered_message) {
auto LogDataMessage = Getf144_LogData(Message.data());
Timestamp.appendElement(LogDataMessage->timestamp());
auto Type = LogDataMessage->value_type();
@@ -273,6 +274,7 @@ void f144_Writer::writeImpl(FlatbufferMessage const &Message) {
MetaDataMax.setValue(Max);
MetaDataMean.setValue(Sum / TotalNrOfElementsWritten);
}
+ return true;
}
void f144_Writer::register_meta_data(hdf5::node::Group const &HDFGroup,
diff --git a/src/WriterModule/f144/f144_Writer.h b/src/WriterModule/f144/f144_Writer.h
index 1841d7941f0c57214f819aaf324efcf87b4a8ca2..b051a739ac0415344cf716d82fd54ffef2371138 100644
--- a/src/WriterModule/f144/f144_Writer.h
+++ b/src/WriterModule/f144/f144_Writer.h
@@ -39,7 +39,8 @@ public:
WriterModule::InitResult reopen(hdf5::node::Group &HDFGroup) override;
/// Write an incoming message which should contain a flatbuffer.
- void writeImpl(FlatbufferMessage const &Message) override;
+ bool writeImpl(FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
f144_Writer()
: WriterModule::Base("f144", false, "NXlog",
diff --git a/src/WriterModule/se00/se00_Writer.cpp b/src/WriterModule/se00/se00_Writer.cpp
index 0880a4892458d7fa6609aa228aebb28ec0a9bddf..0b0bd75d044e4929a7320be0155b69671dc844a4 100644
--- a/src/WriterModule/se00/se00_Writer.cpp
+++ b/src/WriterModule/se00/se00_Writer.cpp
@@ -132,7 +132,8 @@ void msgTypeIsConfigType(se00_Writer::Type ConfigType, ValueUnion MsgType) {
}
}
-void se00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
+bool se00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message,
+ [[maybe_unused]] bool is_buffered_message) {
auto FbPointer = Getse00_SampleEnvironmentData(Message.data());
auto CueIndexValue = Value->current_size();
auto ValuesType = FbPointer->values_type();
@@ -199,7 +200,7 @@ void se00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
Logger::Info("Unknown data type in flatbuffer.");
}
if (NrOfElements == 0) {
- return;
+ return false;
}
CueTimestampIndex.appendElement(static_cast<std::uint32_t>(CueIndexValue));
CueTimestamp.appendElement(FbPointer->packet_timestamp());
@@ -216,6 +217,7 @@ void se00_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
FbPointer->packet_timestamp(), FbPointer->time_delta(), NrOfElements));
Timestamp.appendArray(TempTimeStamps);
}
+ return true;
}
template <typename Type>
diff --git a/src/WriterModule/se00/se00_Writer.h b/src/WriterModule/se00/se00_Writer.h
index 43a770d139439139a12c180000991f2821049751..290c177190b1a9f23c989bf5c0c60e42eede4349 100644
--- a/src/WriterModule/se00/se00_Writer.h
+++ b/src/WriterModule/se00/se00_Writer.h
@@ -44,7 +44,8 @@ public:
InitResult reopen(hdf5::node::Group &HDFGroup) override;
- void writeImpl(FlatbufferMessage const &Message) override;
+ bool writeImpl(FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
enum class Type {
int8,
diff --git a/src/WriterModule/tdct/tdct_Writer.cpp b/src/WriterModule/tdct/tdct_Writer.cpp
index fa0551a64c57edc07a0672f7dd19ea852b020083..2f69c1c7a8f77060e4de34066fb7b0619fa9451c 100644
--- a/src/WriterModule/tdct/tdct_Writer.cpp
+++ b/src/WriterModule/tdct/tdct_Writer.cpp
@@ -64,20 +64,22 @@ WriterModule::InitResult tdct_Writer::reopen(hdf5::node::Group &HDFGroup) {
return WriterModule::InitResult::OK;
}
-void tdct_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message) {
+bool tdct_Writer::writeImpl(const FileWriter::FlatbufferMessage &Message,
+ [[maybe_unused]] bool is_buffered_message) {
auto FbPointer = Gettimestamp(Message.data());
auto TempTimePtr = FbPointer->timestamps()->data();
auto TempTimeSize = FbPointer->timestamps()->size();
if (TempTimeSize == 0) {
Logger::Info(
"Received a flatbuffer with zero (0) timestamps elements in it.");
- return;
+ return false;
}
hdf5::ArrayAdapter<const std::uint64_t> CArray(TempTimePtr, TempTimeSize);
auto CueIndexValue = Timestamp.current_size();
CueTimestampIndex.appendElement(static_cast<std::uint32_t>(CueIndexValue));
CueTimestamp.appendElement(FbPointer->timestamps()->operator[](0));
Timestamp.appendArray(CArray);
+ return true;
}
} // namespace WriterModule::tdct
diff --git a/src/WriterModule/tdct/tdct_Writer.h b/src/WriterModule/tdct/tdct_Writer.h
index d82d770b342e4fad1fb49687264c010632961926..ee6c2c87d1586d1acf7a97bb166fd48dea7f9433 100644
--- a/src/WriterModule/tdct/tdct_Writer.h
+++ b/src/WriterModule/tdct/tdct_Writer.h
@@ -34,7 +34,8 @@ public:
InitResult reopen(hdf5::node::Group &HDFGroup) override;
- void writeImpl(FlatbufferMessage const &Message) override;
+ bool writeImpl(FlatbufferMessage const &Message,
+ bool is_buffered_message) override;
protected:
NeXusDataset::Time Timestamp;
diff --git a/src/WriterModule/template/TemplateWriter.h b/src/WriterModule/template/TemplateWriter.h
index 867e15f3d48ec73688e4fe223c46eb60c166bc07..ea5cae5b3bb4b149a41c735cd9353381a52d1e69 100644
--- a/src/WriterModule/template/TemplateWriter.h
+++ b/src/WriterModule/template/TemplateWriter.h
@@ -186,8 +186,9 @@ public:
/// \param Message The structure containing a pointer to a buffer
/// containing data received from the Kafka broker and the size of the buffer.
// cppcheck-suppress functionStatic
- void writeImpl(FileWriter::FlatbufferMessage const &/*Message*/) override {
+ bool writeImpl([[maybe_unused]] FileWriter::FlatbufferMessage const &message, [[maybe_unused]]bool is_buffered_message) override {
std::cout << "WriterClass::writeImpl()\n";
+ return true;
}
};
} // namespace TemplateWriter
diff --git a/src/WriterModuleBase.h b/src/WriterModuleBase.h
index b0f98ddc3096a23ba56dcf26c4df6d324aabc50d..65d8ac703a9060f181548d25fb9cebff967d779b 100644
--- a/src/WriterModuleBase.h
+++ b/src/WriterModuleBase.h
@@ -111,15 +111,18 @@ public:
/// \brief Increment write counter and call the subclass-specific write logic.
///
/// \param msg The message to process
- void write(FileWriter::FlatbufferMessage const &Message) {
- writeImpl(Message);
- WriteCount++;
+ void write(FileWriter::FlatbufferMessage const &Message,
+ bool is_buffered_message) {
+ if (writeImpl(Message, is_buffered_message)) {
+ WriteCount++;
+ }
}
/// \brief Process the message in some way, for example write to the HDF file.
///
/// \param msg The message to process
- virtual void writeImpl(FileWriter::FlatbufferMessage const &Message) = 0;
+ virtual bool writeImpl(FileWriter::FlatbufferMessage const &Message,
+ bool is_buffered_message) = 0;
void registerField(JsonConfig::FieldBase *Ptr) {
ConfigHandler.registerField(Ptr);
diff --git a/tests/SourceTests.cpp b/tests/SourceTests.cpp
index f2d8c5d30183a87fa6ee7770c3e8f0f667f770c2..49c3a816884869fd159fcf4e80e757d403b45a0e 100644
--- a/tests/SourceTests.cpp
+++ b/tests/SourceTests.cpp
@@ -34,7 +34,7 @@ public:
class WriterModuleMock : public StubWriterModule {
public:
- MAKE_MOCK1(writeImpl, void(FlatbufferMessage const &), override);
+ MAKE_MOCK2(writeImpl, bool(FlatbufferMessage const &, bool), override);
};
TEST_F(SourceTests, ConstructorSetsMembers) {
diff --git a/tests/Stream/MessageWriterTests.cpp b/tests/Stream/MessageWriterTests.cpp
index bff7af812a3fe002bba713a13ef89819f76d1b43..0cec2c971d9a26cda618e2ff9865b948193cb16a 100644
--- a/tests/Stream/MessageWriterTests.cpp
+++ b/tests/Stream/MessageWriterTests.cpp
@@ -25,7 +25,8 @@ public:
~WriterModuleStandIn() = default;
MAKE_MOCK1(init_hdf, WriterModule::InitResult(hdf5::node::Group &), override);
MAKE_MOCK1(reopen, WriterModule::InitResult(hdf5::node::Group &), override);
- MAKE_MOCK1(writeImpl, void(FileWriter::FlatbufferMessage const &), override);
+ MAKE_MOCK2(writeImpl, bool(FileWriter::FlatbufferMessage const &, bool),
+ override);
};
class DataMessageWriterTest : public ::testing::Test {
@@ -79,7 +80,7 @@ TEST_F(DataMessageWriterTest, EnableExtraModule) {
}
TEST_F(DataMessageWriterTest, WriteMessageSuccess) {
- REQUIRE_CALL(WriterModule, writeImpl(_)).TIMES(1);
+ REQUIRE_CALL(WriterModule, writeImpl(_, _)).TIMES(1).RETURN(true);
auto InitialWriteCount = WriterModule.getWriteCount();
FileWriter::FlatbufferMessage Msg;
Stream::Message SomeMessage(
@@ -87,7 +88,7 @@ TEST_F(DataMessageWriterTest, WriteMessageSuccess) {
{
Stream::MessageWriter Writer{
[]() {}, 1s, std::make_unique<Metrics::Registrar>("some_prefix")};
- Writer.addMessage(SomeMessage);
+ Writer.addMessage(SomeMessage, false);
Writer.runJob([&Writer]() {
EXPECT_TRUE(Writer.nrOfWritesDone() == 1);
EXPECT_TRUE(Writer.nrOfWriteErrors() == 0);
@@ -97,7 +98,7 @@ TEST_F(DataMessageWriterTest, WriteMessageSuccess) {
}
TEST_F(DataMessageWriterTest, WriteMessageExceptionUnknownFb) {
- REQUIRE_CALL(WriterModule, writeImpl(_))
+ REQUIRE_CALL(WriterModule, writeImpl(_, _))
.TIMES(1)
.THROW(WriterModule::WriterException("Some error."));
auto InitialWriteCount = WriterModule.getWriteCount();
@@ -110,7 +111,7 @@ TEST_F(DataMessageWriterTest, WriteMessageExceptionUnknownFb) {
Writer.runJob([&Writer]() {
EXPECT_TRUE(Writer.nrOfWriterModulesWithErrors() == 1);
});
- Writer.addMessage(SomeMessage);
+ Writer.addMessage(SomeMessage, false);
Writer.runJob([&Writer]() {
EXPECT_TRUE(Writer.nrOfWritesDone() == 0);
EXPECT_TRUE(Writer.nrOfWriteErrors() == 1);
diff --git a/tests/Stream/SourceFilterTest.cpp b/tests/Stream/SourceFilterTest.cpp
index 1f4d0c562dfc740a3617c1d4730612687be3c247..72b3f0a5ca99d90fb9c063765f7de7973f382612 100644
--- a/tests/Stream/SourceFilterTest.cpp
+++ b/tests/Stream/SourceFilterTest.cpp
@@ -17,8 +17,9 @@ class StubMessageWriter : public Stream::MessageWriter {
public:
StubMessageWriter()
: MessageWriter([]() {}, 1s, std::make_unique<Metrics::Registrar>("")) {}
- void addMessage(Stream::Message const &Msg) override {
- messages_received.emplace_back(Msg);
+ void addMessage(Stream::Message const &message,
+ [[maybe_unused]] bool is_buffered_message) override {
+ messages_received.emplace_back(message);
}
std::vector<Stream::Message> messages_received;
};
@@ -115,15 +116,13 @@ TEST(SourceFilter, message_on_start_is_allowed_through) {
EXPECT_EQ(1u, harness.writer->messages_received.size());
}
-TEST(SourceFilter, first_message_after_stop_is_allowed_through) {
+TEST(SourceFilter, messages_after_stop_are_not_allowed_through) {
auto harness = create_filter_for_tests(time_point{0ms}, time_point{1000ms});
harness.filter->filter_message(create_f144_message("::source::", 1, 1001));
harness.filter->filter_message(create_f144_message("::source::", 2, 1002));
- EXPECT_EQ(1u, harness.writer->messages_received.size());
- auto first = harness.writer->messages_received.at(0).FbMsg;
- EXPECT_EQ(1001000000, first.getTimestamp()); // timestamp is in ns
+ EXPECT_EQ(0u, harness.writer->messages_received.size());
}
TEST(SourceFilter, message_after_stop_sets_filter_to_finished) {
@@ -163,11 +162,13 @@ TEST(SourceFilter, can_change_stop_time_after_construction) {
harness.filter->set_stop_time(time_point{1000ms});
- // First message after stop is allowed
+ // Messages after stop are not allowed
+ harness.filter->filter_message(create_f144_message("::source::", 1, 999));
+ harness.filter->filter_message(create_f144_message("::source::", 1, 1000));
harness.filter->filter_message(create_f144_message("::source::", 1, 1001));
harness.filter->filter_message(create_f144_message("::source::", 2, 1002));
- EXPECT_EQ(1u, harness.writer->messages_received.size());
+ EXPECT_EQ(2u, harness.writer->messages_received.size());
}
TEST(SourceFilter,
diff --git a/tests/WriterModule/TemplateWriterTests.cpp b/tests/WriterModule/TemplateWriterTests.cpp
index 09f9e20374bf1cb39802506570c680d926ab2859..f23a9e7b6912b8139787c50fecaa628ca21de9c9 100644
--- a/tests/WriterModule/TemplateWriterTests.cpp
+++ b/tests/WriterModule/TemplateWriterTests.cpp
@@ -15,5 +15,5 @@ TEST(TemplateTests, WriterReturnValues) {
hdf5::node::Group SomeGroup;
EXPECT_TRUE(SomeWriter.init_hdf(SomeGroup) == WriterModule::InitResult::OK);
EXPECT_TRUE(SomeWriter.reopen(SomeGroup) == WriterModule::InitResult::OK);
- EXPECT_NO_THROW(SomeWriter.write(FileWriter::FlatbufferMessage()));
+ EXPECT_NO_THROW(SomeWriter.write(FileWriter::FlatbufferMessage(), false));
}
diff --git a/tests/WriterModule/al00_WriterTests.cpp b/tests/WriterModule/al00_WriterTests.cpp
index d0c103b994228e3681522893f771853561a70891..5e9f6d9c81c689cf1e964e1c24c7fe04b8ee59cd 100644
--- a/tests/WriterModule/al00_WriterTests.cpp
+++ b/tests/WriterModule/al00_WriterTests.cpp
@@ -89,7 +89,7 @@ TEST_F(EPICS_AlarmWriter, WriteDataOnce) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto AlarmMsgDataset = UsedGroup.get_dataset("alarm_message");
auto AlarmSeverityDataset = UsedGroup.get_dataset("alarm_severity");
auto AlarmTimeDataset = UsedGroup.get_dataset("alarm_time");
diff --git a/tests/WriterModule/da00_WriterTests.cpp b/tests/WriterModule/da00_WriterTests.cpp
index c37d0bc7cbb1e3f9f335e2e378adb94e6f1d7786..3d46bfe675b08b82d879ac21ba9394e48f38a942 100644
--- a/tests/WriterModule/da00_WriterTests.cpp
+++ b/tests/WriterModule/da00_WriterTests.cpp
@@ -747,7 +747,7 @@ TEST_F(da00_WriterTestFixture, da00_WriterWriteExtendsShape) {
writer.init_hdf(_group);
writer.reopen(_group);
for (int i = 0; i < write_count; ++i)
- EXPECT_NO_THROW(writer.write(message));
+ EXPECT_NO_THROW(writer.write(message, false));
EXPECT_EQ(write_count, writer.Timestamp.current_size());
}
for (const auto &name :
@@ -774,7 +774,7 @@ TEST_F(da00_WriterTestFixture, da00_WriterWriteCopiesData) {
writer.reopen(_group);
constexpr int write_count{10};
for (int i = 0; i < write_count; ++i)
- EXPECT_NO_THROW(writer.write(message));
+ EXPECT_NO_THROW(writer.write(message, false));
EXPECT_EQ(write_count, writer.Timestamp.current_size());
}
const auto shape = Simple(_group.get_dataset("signal").dataspace());
@@ -797,7 +797,7 @@ TEST_F(da00_WriterTestFixture, da00_WriterFillsInAttributes) {
writer.reopen(_group);
constexpr int write_count{10};
for (int i = 0; i < write_count; ++i)
- EXPECT_NO_THROW(writer.write(message));
+ EXPECT_NO_THROW(writer.write(message, false));
EXPECT_EQ(write_count, writer.Timestamp.current_size());
}
for (const auto &name : {"signal", "signal_error", "gain", "x", "y", "time",
diff --git a/tests/WriterModule/ep01_WriterTests.cpp b/tests/WriterModule/ep01_WriterTests.cpp
index 088bf7c2cb7428babb18c8fdbd9e0446e73bd763..40664fc91c9953d9b386643d7cf7b8b8e41ae813 100644
--- a/tests/WriterModule/ep01_WriterTests.cpp
+++ b/tests/WriterModule/ep01_WriterTests.cpp
@@ -87,7 +87,7 @@ TEST_F(EPICS_ConStatusWriter, WriteDataOnce) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto ConStatusTimeDataset = UsedGroup.get_dataset("connection_status_time");
auto ConStatusDataset = UsedGroup.get_dataset("connection_status");
auto FbPointer = GetEpicsPVConnectionInfo(TestMsg.data());
diff --git a/tests/WriterModule/ev44_WriterTests.cpp b/tests/WriterModule/ev44_WriterTests.cpp
index 748f7de19523af2ae4ee894be7cfebbf6141bf2c..a3b904fc3499e268678c71c76b9f3b720d468ce6 100644
--- a/tests/WriterModule/ev44_WriterTests.cpp
+++ b/tests/WriterModule/ev44_WriterTests.cpp
@@ -163,7 +163,7 @@ TEST_F(Event44WriterTests,
WriterModule::ev44::ev44_Writer Writer;
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage));
+ EXPECT_NO_THROW(Writer.write(TestMessage, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -217,7 +217,7 @@ TEST_F(Event44WriterTests, WriterSuccessfullyRecordsEventDataWithoutPixelIds) {
WriterModule::ev44::ev44_Writer Writer;
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage));
+ EXPECT_NO_THROW(Writer.write(TestMessage, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -280,8 +280,8 @@ TEST_F(Event44WriterTests, WriterSuccessfullyRecordsEventDataFromTwoMessages) {
WriterModule::ev44::ev44_Writer Writer;
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage1));
- EXPECT_NO_THROW(Writer.write(TestMessage2));
+ EXPECT_NO_THROW(Writer.write(TestMessage1, false));
+ EXPECT_NO_THROW(Writer.write(TestMessage2, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -369,9 +369,9 @@ TEST_F(Event44WriterTests, WriterSuccessfullyHandlesMessageWithNoEvents) {
WriterModule::ev44::ev44_Writer Writer;
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage1));
- EXPECT_NO_THROW(Writer.write(TestMessage2));
- EXPECT_NO_THROW(Writer.write(TestMessage3));
+ EXPECT_NO_THROW(Writer.write(TestMessage1, false));
+ EXPECT_NO_THROW(Writer.write(TestMessage2, false));
+ EXPECT_NO_THROW(Writer.write(TestMessage3, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -445,7 +445,7 @@ TEST_F(Event44WriterTests, PulseTimeIsRepeatedWithinSameMessage) {
WriterModule::ev44::ev44_Writer Writer;
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage));
+ EXPECT_NO_THROW(Writer.write(TestMessage, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -510,8 +510,8 @@ TEST_F(Event44WriterTests, LastPulseTimeIsRepeatedInSubsequentMessage) {
WriterModule::ev44::ev44_Writer Writer;
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage1));
- EXPECT_NO_THROW(Writer.write(TestMessage2));
+ EXPECT_NO_THROW(Writer.write(TestMessage1, false));
+ EXPECT_NO_THROW(Writer.write(TestMessage2, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -581,8 +581,8 @@ TEST_F(Event44WriterTests, CuesFromTwoMessagesAreRecorded) {
Writer.setCueInterval(1);
EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
- EXPECT_NO_THROW(Writer.write(TestMessage1));
- EXPECT_NO_THROW(Writer.write(TestMessage2));
+ EXPECT_NO_THROW(Writer.write(TestMessage1, false));
+ EXPECT_NO_THROW(Writer.write(TestMessage2, false));
} // These braces are required due to "h5.cpp"
// Read data from the file
@@ -603,3 +603,30 @@ TEST_F(Event44WriterTests, CuesFromTwoMessagesAreRecorded) {
<< "Expected cue_index dataset to contain the indices of the last event "
"of every message";
}
+
+TEST_F(Event44WriterTests, buffered_data_not_written) {
+ std::vector<int32_t> TimeOfFlight1 = {101, 102, 201};
+ std::vector<int32_t> DetectorID1 = {101, 102, 201};
+ std::vector<int64_t> ReferenceTime1 = {1000, 2000};
+ std::vector<int32_t> ReferenceTimeIndex1 = {0, 2};
+ auto MessageBuffer1 =
+ generateFlatbufferData("TestSource", 1, TimeOfFlight1, DetectorID1,
+ ReferenceTime1, ReferenceTimeIndex1);
+ FileWriter::FlatbufferMessage TestMessage1(MessageBuffer1.data(),
+ MessageBuffer1.size());
+
+ // Create writer and give it the message to write
+ {
+ WriterModule::ev44::ev44_Writer Writer;
+ EXPECT_TRUE(Writer.init_hdf(TestGroup) == InitResult::OK);
+ EXPECT_TRUE(Writer.reopen(TestGroup) == InitResult::OK);
+ EXPECT_NO_THROW(Writer.write(TestMessage1, true));
+ } // These braces are required due to "h5.cpp"
+
+ // Read data from the file
+ auto EventTimeOffsetDataset = TestGroup.get_dataset("event_time_offset");
+ auto EventTimeZeroDataset = TestGroup.get_dataset("event_time_zero");
+
+ EXPECT_EQ(0, EventTimeOffsetDataset.dataspace().size());
+ EXPECT_EQ(0, EventTimeZeroDataset.dataspace().size());
+}
diff --git a/tests/WriterModule/f144_WriterTests.cpp b/tests/WriterModule/f144_WriterTests.cpp
index 05b73f29887fd0f0dd3e2c4f5143c9dd87548e67..0948b4c32d33e3e66a4e632eefff8320ccaa1364 100644
--- a/tests/WriterModule/f144_WriterTests.cpp
+++ b/tests/WriterModule/f144_WriterTests.cpp
@@ -313,7 +313,7 @@ TEST_F(f144Init, write_one_element) {
EXPECT_EQ(FlatbufferMsg.getFlatbufferID(), "f144");
EXPECT_EQ(TestWriter.Values.size(), 0u);
EXPECT_EQ(TestWriter.Timestamp.current_size(), 0);
- TestWriter.write(FlatbufferMsg);
+ TestWriter.write(FlatbufferMsg, false);
EXPECT_EQ(TestWriter.Values.size(), 1u);
ASSERT_EQ(TestWriter.Timestamp.current_size(), 1);
std::vector<double> WrittenValues(1);
@@ -338,7 +338,8 @@ TEST_F(f144Init, write_one_default_value_element) {
EXPECT_EQ(TestWriter.Values.size(), 0u);
EXPECT_EQ(TestWriter.Timestamp.current_size(), 0);
TestWriter.write(FileWriter::FlatbufferMessage(FlatbufferData.first.get(),
- FlatbufferData.second));
+ FlatbufferData.second),
+ false);
EXPECT_EQ(TestWriter.Values.size(), 1u);
ASSERT_EQ(TestWriter.Timestamp.current_size(), 1);
std::vector<double> WrittenValues(1);
@@ -360,7 +361,7 @@ TEST_F(f144Init, write_multiple_elements) {
auto [buffer, size] =
f144_schema::generateFlatbufferMessage(values[i], timestamps[i]);
FileWriter::FlatbufferMessage FlatbufferMsg(buffer.get(), size);
- TestWriter.write(FlatbufferMsg);
+ TestWriter.write(FlatbufferMsg, false);
}
std::vector<double> WrittenValues(values.size());
diff --git a/tests/WriterModule/se00_WriterTests.cpp b/tests/WriterModule/se00_WriterTests.cpp
index b7a663454b4de3af2356660b11989d4a12e9ea0d..74bb73a92946078b3feac104cfa98ed1521e210d 100644
--- a/tests/WriterModule/se00_WriterTests.cpp
+++ b/tests/WriterModule/se00_WriterTests.cpp
@@ -109,7 +109,7 @@ TEST_F(se00Writer, WriteDataOnce) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto RawValuesDataset = UsedGroup.get_dataset("value");
auto TimestampDataset = UsedGroup.get_dataset("time");
auto CueIndexDataset = UsedGroup.get_dataset("cue_index");
@@ -151,8 +151,8 @@ TEST_F(se00Writer, WriteDataTwice) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto RawValuesDataset = UsedGroup.get_dataset("value");
auto TimestampDataset = UsedGroup.get_dataset("time");
auto CueIndexDataset = UsedGroup.get_dataset("cue_index");
@@ -190,7 +190,7 @@ TEST_F(se00Writer, WriteNoElements) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto RawValuesDataset = UsedGroup.get_dataset("value");
auto TimestampDataset = UsedGroup.get_dataset("time");
auto CueIndexDataset = UsedGroup.get_dataset("cue_index");
@@ -209,7 +209,7 @@ TEST_F(se00Writer, WriteDataWithNoTimestampsInFB) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto RawValuesDataset = UsedGroup.get_dataset("value");
auto TimestampDataset = UsedGroup.get_dataset("time");
auto CueIndexDataset = UsedGroup.get_dataset("cue_index");
diff --git a/tests/WriterModule/tdct_WriterTests.cpp b/tests/WriterModule/tdct_WriterTests.cpp
index a80559b55780b0b25e264905ccbcef103e7e1f67..e8d0e8d4146967190b85ea8bf9ba33faae72ddaf 100644
--- a/tests/WriterModule/tdct_WriterTests.cpp
+++ b/tests/WriterModule/tdct_WriterTests.cpp
@@ -89,7 +89,7 @@ TEST_F(ChopperTimeStampWriter, WriteDataOnce) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto TimestampDataset = UsedGroup.get_dataset("time");
auto CueIndexDataset = UsedGroup.get_dataset("cue_index");
auto CueTimestampZeroDataset = UsedGroup.get_dataset("cue_timestamp_zero");
@@ -119,8 +119,8 @@ TEST_F(ChopperTimeStampWriter, WriteDataTwice) {
EXPECT_TRUE(Writer.init_hdf(UsedGroup) == InitResult::OK);
EXPECT_TRUE(Writer.reopen(UsedGroup) == InitResult::OK);
FileWriter::FlatbufferMessage TestMsg(Buffer.get(), BufferSize);
- EXPECT_NO_THROW(Writer.write(TestMsg));
- EXPECT_NO_THROW(Writer.write(TestMsg));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
+ EXPECT_NO_THROW(Writer.write(TestMsg, false));
auto TimestampDataset = UsedGroup.get_dataset("time");
auto CueIndexDataset = UsedGroup.get_dataset("cue_index");
auto CueTimestampZeroDataset = UsedGroup.get_dataset("cue_timestamp_zero");
diff --git a/tests/helpers/StubWriterModule.h b/tests/helpers/StubWriterModule.h
index 0fc753363d69358b52eb66efa520f939fd892ab9..d75dd844d2dfc5733868b0c68b9cfce26c316d06 100644
--- a/tests/helpers/StubWriterModule.h
+++ b/tests/helpers/StubWriterModule.h
@@ -22,5 +22,8 @@ public:
InitResult reopen(hdf5::node::Group & /*HDFGroup*/) override {
return InitResult::OK;
}
- void writeImpl(FileWriter::FlatbufferMessage const & /*Message*/) override {}
+ bool writeImpl([[maybe_unused]] FileWriter::FlatbufferMessage const &message,
+ [[maybe_unused]] bool is_buffered_message) override {
+ return true;
+ }
};