Merge branch 'jfjoch_cbor_in_place' into main

common/FrameTransformation.cpp

@@ -40,11 +40,6 @@ FrameTransformation::FrameTransformation(const DiffractionExperiment &in_experim
     }
 }
 
-FrameTransformation& FrameTransformation::SetOutput(void *output) {
-    standard_output = (char *) output;
-    return *this;
-}
-
 template <class Td> void AddToFramesSum(Td *destination, const int16_t *source) {
     for (int i = 0; i < RAW_MODULE_SIZE; i++) {
         if ((source[i] == INT16_MIN) || (destination[i] == INT32_MIN))
@@ -90,9 +85,7 @@ void FrameTransformation::Generate16BitPreview() {
     }
 }
 
-size_t FrameTransformation::PackStandardOutput() {
-    if (standard_output == nullptr)
-        throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, "Default stream output not initialized");
+void FrameTransformation::Pack() {
 
     if (summation > 1) {
         PackSummation();
@@ -114,7 +107,11 @@ size_t FrameTransformation::PackStandardOutput() {
                                 experiment.GetYPixelsNumFullImage(),
                                 INT16_MIN, INT16_MAX);
     }
-    return compressor.Compress(standard_output, precompression_buffer.data(),
+
+}
+
+size_t FrameTransformation::SaveCompressedImage(void *output) {
+    return compressor.Compress((char *) output, precompression_buffer.data(),
                                experiment.GetPixelsNum(), pixel_depth);
 }
 

common/FrameTransformation.h

@@ -16,8 +16,6 @@ class FrameTransformation {
     std::vector<char> precompression_buffer;
     std::vector<int16_t> image16bit;
 
-    char *standard_output = nullptr;
-
     const size_t summation;
     const size_t pixel_depth;
 
@@ -27,12 +25,11 @@ class FrameTransformation {
     void Generate16BitPreview();
 public:
     FrameTransformation(const DiffractionExperiment &experiment);
-    FrameTransformation& SetOutput(void *output);
     void ProcessModule(const int16_t *input, uint16_t module_number, int data_stream);
     void ProcessModule(JFConversion &conv, const int16_t *input, uint16_t module_number,
                        int data_stream);
-    size_t PackStandardOutput(); // transfer summed image to converted coordinates, clear summation buffer, compress
-
+    void Pack(); // transfer summed image to converted coordinates, clear summation buffer
+    size_t SaveCompressedImage(void *output);
     int16_t *GetPreview16BitImage();
 };
 

common/ImagePusher.cpp

@@ -15,15 +15,3 @@ void PrepareCBORImage(DataMessage& message,
     message.image.algorithm = experiment.GetCompressionAlgorithmEnum();
     message.image.channel = "default";
 }
-
-
-void PrepareDataMessageSpots(DataMessage& message,
-                             const std::vector<DiffractionSpot>& spots) {
-    message.spots.clear();
-    for (const auto & spot : spots)
-        message.spots.push_back(spot);
-}
-
-void ImagePusher::SendData(const std::vector<uint8_t> &serialized_image, int64_t image_number) {
-    SendDataInternal(serialized_image, image_number);
-}

common/ImagePusher.h

@@ -18,12 +18,10 @@ void PrepareCBORImage(DataMessage& message,
                       void *image, size_t image_size);
 
 class ImagePusher {
-protected:
-    virtual void SendDataInternal(const std::vector<uint8_t>& serialized_image,  int64_t image_number) = 0;
 public:
     virtual void StartDataCollection(const StartMessage& message) = 0;
     virtual void EndDataCollection(const EndMessage& message) = 0;
-    void SendData(const std::vector<uint8_t>& serialized_image,  int64_t image_number);
+    virtual void SendImage(const uint8_t *image_data, size_t image_size,  int64_t image_number) = 0;
 };
 
 

common/TestImagePusher.cpp

@@ -29,14 +29,14 @@ void TestImagePusher::EndDataCollection(const EndMessage& message) {
         is_running = false;
 }
 
-void TestImagePusher::SendDataInternal(const std::vector<uint8_t> &serialized_image, int64_t image_number) {
+void TestImagePusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) {
     std::unique_lock<std::mutex> ul(m);
 
     frame_counter++;
     if (image_number == image_id) {
 
         JFJochFrameDeserializer deserializer;
-        deserializer.Process(serialized_image);
+        deserializer.Process(image_data, image_size);
 
         auto image_array = deserializer.GetDataMessage();
         receiver_generated_image.resize(image_array.image.size);

common/TestImagePusher.h

@@ -19,8 +19,8 @@ class TestImagePusher : public ImagePusher {
     bool correct_sequence = true;
     bool is_running = false;
     size_t frame_counter = 0;
-    void SendDataInternal(const std::vector<uint8_t> &serialized_image, int64_t image_number) override;
 public:
+    void SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override;
     explicit TestImagePusher(int64_t image_number);
     void StartDataCollection(const StartMessage& message) override;
     void EndDataCollection(const EndMessage& message) override;

common/ZMQImagePusher.cpp

@@ -40,11 +40,11 @@ ZMQImagePusher::ZMQImagePusher(const std::vector<std::string> &addr,
     }
 }
 
-void ZMQImagePusher::SendDataInternal(const std::vector<uint8_t> &serialized_image, int64_t image_number) {
+void ZMQImagePusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number)  {
     if (sockets.empty())
         return;
     auto socket_number = (image_number % file_count) % sockets.size();
-    sockets[socket_number]->Send(serialized_image.data(), serialized_image.size());
+    sockets[socket_number]->Send(image_data, image_size);
 }
 
 void ZMQImagePusher::StartDataCollection(const StartMessage& message) {
@@ -56,9 +56,9 @@ void ZMQImagePusher::StartDataCollection(const StartMessage& message) {
     file_count = message.data_file_count;
 
     serializer.SerializeSequenceStart(message);
-    auto &buffer = serializer.GetBuffer();
+
     for (const auto &s: sockets)
-        s->Send(buffer.data(), buffer.size(), true);
+        s->Send(serializer.GetBuffer(), serializer.GetBufferSize(), true);
 }
 
 
@@ -69,9 +69,7 @@ void ZMQImagePusher::EndDataCollection(const EndMessage& message) {
 
     for (const auto &s: sockets) {
         serializer.SerializeSequenceEnd(end_message);
-        auto &buffer = serializer.GetBuffer();
-
-        s->Send(buffer.data(), buffer.size(), true);
+        s->Send(serializer.GetBuffer(), serializer.GetBufferSize(), true);
         end_message.write_master_file = false;
     }
 }

common/ZMQImagePusher.h

@@ -16,8 +16,8 @@ class ZMQImagePusher : public ImagePusher {
     std::vector<std::unique_ptr<ZMQContext>> contexts;
     std::vector<std::unique_ptr<ZMQSocket>> sockets;
     int64_t file_count = 1;
-    void SendDataInternal(const std::vector<uint8_t>& serialized_image,  int64_t image_number) override;
 public:
+    void SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) override;
     ZMQImagePusher(ZMQContext &context, const std::vector<std::string>& addr,
                    int32_t send_buffer_high_watermark = -1, int32_t send_buffer_size = -1);
     // High performance implementation, where each socket has dedicated ZMQ context

frame_serialize/JFJochFrameDeserializer.cpp

@@ -603,12 +603,16 @@ void JFJochFrameDeserializer::ProcessImageData(CborValue &value) {
 }
 
 void JFJochFrameDeserializer::Process(const std::vector<uint8_t> &buffer) {
+    Process(buffer.data(), buffer.size());
+}
+
+void JFJochFrameDeserializer::Process(const uint8_t *msg, size_t msg_size) {
 std::unique_lock<std::mutex> ul(m);
     data_message = DataMessage();
 
     CborParser parser;
     CborValue value;
-    cborErr(cbor_parser_init(buffer.data(), buffer.size(), 0, &parser, &value));
+    cborErr(cbor_parser_init(msg, msg_size, 0, &parser, &value));
 
     if (GetCBORTag(value) != CborSignatureTag)
         throw JFJochException(JFJochExceptionCategory::CBORError, "CBOR must start with dedicated tag");

frame_serialize/JFJochFrameDeserializer.h

@@ -47,6 +47,7 @@ private:
     bool ProcessEndMessageElement(CborValue &value);
 public:
     void Process(const std::vector<uint8_t>& buffer);
+    void Process(const uint8_t *msg, size_t msg_size);
     [[nodiscard]] Type GetType() const;
     [[nodiscard]] EndMessage GetEndMessage() const;
     [[nodiscard]] StartMessage GetStartMessage() const;

frame_serialize/JFJochFrameSerializer.cpp

@@ -52,9 +52,9 @@ void CBOR_ENC_COMPRESSED(CborEncoder &encoder,
                          const void *image, size_t image_size,
                          CompressionAlgorithm algorithm,
                          size_t elem_size) {
-    if (algorithm == CompressionAlgorithm::NO_COMPRESSION) {
+    if (algorithm == CompressionAlgorithm::NO_COMPRESSION)
         cborErr(cbor_encode_byte_string(&encoder, (uint8_t *) image, image_size));
-    } else {
+    else {
         cbor_encode_tag(&encoder, TagDECTRISCompression);
 
         CborEncoder arrayEncoder;
@@ -82,8 +82,8 @@ inline void CBOR_ENC_MULTIDIM_TYPED_ARRAY(CborEncoder &encoder, const char* key,
                                               const void *image, size_t image_size,
                                               size_t xpixel, size_t ypixel, CompressionAlgorithm algorithm,
                                               size_t elem_size, bool elem_sign) {
-    if ((algorithm == CompressionAlgorithm::NO_COMPRESSION) && (xpixel * ypixel != image_size / elem_size))
-        throw JFJochException(JFJochExceptionCategory::CBORError, "Mismatch in array size");
+    //if ((algorithm == CompressionAlgorithm::NO_COMPRESSION) && (xpixel * ypixel != image_size / elem_size))
+    //    throw JFJochException(JFJochExceptionCategory::CBORError, "Mismatch in array size");
 
     CborEncoder arrayEncoder, arrayEncoder_2;
     cborErr(cbor_encode_text_stringz(&encoder, key));
@@ -288,16 +288,44 @@ inline void CBOR_ENC_USER_DATA(CborEncoder &encoder, const StartMessage& message
 }
 
 JFJochFrameSerializer::JFJochFrameSerializer(size_t in_max_buffer_size) :
-max_buffer_size(in_max_buffer_size) {
-    buffer.reserve(in_max_buffer_size);
+max_buffer_size(in_max_buffer_size), curr_size(0) {
+    buffer.resize(max_buffer_size);
 }
 
-const std::vector<uint8_t> &JFJochFrameSerializer::GetBuffer() const {
-    return buffer;
+const uint8_t *JFJochFrameSerializer::GetBuffer() const {
+    return buffer.data();
 }
 
+uint8_t *JFJochFrameSerializer::GetImageAppendLocation() {
+    return buffer.data() + curr_size + sizeof(size_t);
+}
+
+size_t JFJochFrameSerializer::GetBufferSize() const {
+    return curr_size;
+}
+
+size_t JFJochFrameSerializer::GetRemainingBuffer() const {
+    return max_buffer_size - curr_size;
+}
+
+void JFJochFrameSerializer::AppendImage(size_t image_size) {
+    if (curr_size + image_size + sizeof(size_t) >= max_buffer_size)
+        throw JFJochException(JFJochExceptionCategory::CBORError, "No space to extend the image");
+    curr_size--;
+    buffer[curr_size] = 0x40 | 27;
+    curr_size++;
+#ifdef LITTLE_ENDIAN
+    size_t image_size_be = __builtin_bswap64(image_size);
+#else
+    size_t image_size_be = image_size;
+#endif
+    memcpy(buffer.data() + curr_size, &image_size_be, sizeof(size_t));
+    curr_size += sizeof(size_t);
+    curr_size += image_size;
+}
+
+
 void JFJochFrameSerializer::SerializeSequenceStart(const StartMessage& message) {
-    buffer.resize(max_buffer_size);
     CborEncoder encoder, mapEncoder;
 
     cbor_encoder_init(&encoder, buffer.data(), buffer.size(), 0);
@@ -342,13 +370,10 @@ void JFJochFrameSerializer::SerializeSequenceStart(const StartMessage& message)
     CBOR_ENC_CHANNELS(mapEncoder, "channels", message.channels);
 
     cborErr(cbor_encoder_close_container(&encoder, &mapEncoder));
-
-    buffer.resize(cbor_encoder_get_buffer_size(&encoder, buffer.data()));
+    curr_size = cbor_encoder_get_buffer_size(&encoder, buffer.data());
 }
 
 void JFJochFrameSerializer::SerializeSequenceEnd(const EndMessage& message) {
-    buffer.resize(max_buffer_size);
-
     CborEncoder encoder, mapEncoder;
     cbor_encoder_init(&encoder, buffer.data(), buffer.size(), 0);
     cborErr(cbor_encode_tag(&encoder,CborSignatureTag ));
@@ -365,12 +390,10 @@ void JFJochFrameSerializer::SerializeSequenceEnd(const EndMessage& message) {
 
     cborErr(cbor_encoder_close_container(&encoder, &mapEncoder));
 
-    buffer.resize(cbor_encoder_get_buffer_size(&encoder, buffer.data()));
+    curr_size = cbor_encoder_get_buffer_size(&encoder, buffer.data());
 }
 
 void JFJochFrameSerializer::SerializeImage(const DataMessage& message) {
-    buffer.resize(max_buffer_size);
-
     CborEncoder encoder, mapEncoder, userDataMapEncoder;
     cbor_encoder_init(&encoder, buffer.data(), buffer.size(), 0);
 
@@ -397,5 +420,6 @@ void JFJochFrameSerializer::SerializeImage(const DataMessage& message) {
 
     cborErr(cbor_encoder_close_container(&encoder, &mapEncoder));
 
-    buffer.resize(cbor_encoder_get_buffer_size(&encoder, buffer.data()));
+    curr_size = cbor_encoder_get_buffer_size(&encoder, buffer.data());
 }
+

frame_serialize/JFJochFrameSerializer.h

@@ -15,12 +15,18 @@
 class JFJochFrameSerializer {
     std::vector<uint8_t> buffer;
     size_t max_buffer_size;
+    size_t curr_size;
 public:
     explicit JFJochFrameSerializer(size_t buffer_size);
-    [[nodiscard]] const std::vector<uint8_t> &GetBuffer() const;
+    [[nodiscard]] const uint8_t *GetBuffer() const;
+    [[nodiscard]] uint8_t *GetImageAppendLocation();
+    [[nodiscard]] size_t GetBufferSize() const;
+    [[nodiscard]] size_t GetRemainingBuffer() const;
     void SerializeSequenceStart(const StartMessage& message);
     void SerializeSequenceEnd(const EndMessage& message);
     void SerializeImage(const DataMessage& message);
+    void AppendImage(size_t image_size);
+
 };
 
 

receiver/JFJochReceiver.cpp

@@ -322,8 +322,6 @@ int64_t JFJochReceiver::FrameTransformationThread() {
     int64_t max_thread_delay = 0;
     uint64_t image_number;
 
-    transformation.SetOutput(writer_buffer.data());
-
     frame_transformation_ready->CountDown();
 
     std::unique_ptr<IndexerWrapper> indexer;
@@ -403,7 +401,7 @@ int64_t JFJochReceiver::FrameTransformationThread() {
             }
 
             if (send_image) {
-                size_t image_size = transformation.PackStandardOutput();
+                transformation.Pack();
 
                 std::vector<DiffractionSpot> spots;
                 auto local_data_processing_settings = GetDataProcessingSettings();
@@ -460,13 +458,15 @@ int64_t JFJochReceiver::FrameTransformationThread() {
                 }
 
                 if (push_images_to_writer) {
-                    PrepareCBORImage(message, experiment, writer_buffer.data(), image_size);
+                    PrepareCBORImage(message, experiment, nullptr, 0);
                     serializer.SerializeImage(message);
-                    image_pusher.SendData(serializer.GetBuffer(), image_number);
+                    size_t image_size = transformation.SaveCompressedImage(serializer.GetImageAppendLocation());
+                    serializer.AppendImage(image_size);
+                    image_pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), image_number);
+                    compressed_size += image_size;
                 }
 
                 UpdateMaxImage(image_number);
-                compressed_size += image_size;
                 images_sent++;
             }
         } catch (const JFJochException &e) { Abort(e); }

tests/CBORTest.cpp

@@ -63,10 +63,8 @@ TEST_CASE("CBORSerialize_Start", "[CBOR]") {
 
     REQUIRE_NOTHROW(serializer.SerializeSequenceStart(message));
 
-    auto image = serializer.GetBuffer();
-
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(image));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::START);
 
     StartMessage output_message;
@@ -137,11 +135,9 @@ TEST_CASE("CBORSerialize_End", "[CBOR]") {
     };
 
     REQUIRE_NOTHROW(serializer.SerializeSequenceEnd(message));
-
-    auto image = serializer.GetBuffer();
-
+    
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(image));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::END);
 
     EndMessage output_message{};
@@ -188,10 +184,9 @@ TEST_CASE("CBORSerialize_Image", "[CBOR]") {
     };
 
     REQUIRE_NOTHROW(serializer.SerializeImage(message));
-    auto serialized = serializer.GetBuffer();
 
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(serialized));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
 
     auto image_array = deserializer.GetDataMessage();
@@ -241,12 +236,58 @@ TEST_CASE("CBORSerialize_Image_2", "[CBOR]") {
     };
 
     REQUIRE_NOTHROW(serializer.SerializeImage(message));
-    auto serialized = serializer.GetBuffer();
 
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(serialized));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
 
+    auto image_array = deserializer.GetDataMessage();
+    REQUIRE(image_array.image.algorithm == CompressionAlgorithm::NO_COMPRESSION);
+    REQUIRE(image_array.image.xpixel == 1024);
+    REQUIRE(image_array.image.ypixel == 512);
+    REQUIRE(image_array.image.pixel_depth_bytes == 1);
+    REQUIRE(!image_array.image.pixel_is_signed);
+    REQUIRE(image_array.image.channel == "default");
+    REQUIRE(image_array.image.size == test.size());
+    REQUIRE(image_array.indexing_result == message.indexing_result);
+    REQUIRE(image_array.number == 480);
+    REQUIRE(memcmp(image_array.image.data, test.data(), test.size()) == 0);
+}
+
+TEST_CASE("CBORSerialize_Image_Append", "[CBOR]") {
+    JFJochFrameSerializer serializer(8*1024*1024);
+
+    std::vector<SpotToSave> spots;
+
+    std::vector<uint8_t> test(512*1024);
+    for (int i = 0; i < test.size(); i++)
+        test[i] = (i * 253 + 56) % 256;
+
+    CBORImage image {
+            .data = nullptr,
+            .size = 0,
+            .xpixel = 1024,
+            .ypixel = 512,
+            .pixel_depth_bytes = 1,
+            .pixel_is_signed = false,
+            .algorithm = CompressionAlgorithm::NO_COMPRESSION,
+            .channel = "default"
+    };
+
+    DataMessage message {
+            .number = 480,
+            .image = image,
+            .spots = spots,
+            .indexing_result = 3
+    };
+
+    REQUIRE_NOTHROW(serializer.SerializeImage(message));
+    memcpy(serializer.GetImageAppendLocation(), test.data(), 512*1024);
+    REQUIRE_NOTHROW(serializer.AppendImage(512*1024));
+
+    JFJochFrameDeserializer deserializer;
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
+    REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
 
     auto image_array = deserializer.GetDataMessage();
     REQUIRE(image_array.image.algorithm == CompressionAlgorithm::NO_COMPRESSION);
@@ -288,10 +329,9 @@ TEST_CASE("CBORSerialize_Image_Compressed", "[CBOR]") {
     };
 
     REQUIRE_NOTHROW(serializer.SerializeImage(message));
-    auto serialized = serializer.GetBuffer();
 
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(serialized));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
 
     auto image_array = deserializer.GetDataMessage();
@@ -331,10 +371,8 @@ TEST_CASE("CBORSerialize_Image_Rad_Int_Profile", "[CBOR]") {
 
     REQUIRE_NOTHROW(serializer.SerializeImage(message));
 
-    auto serialized = serializer.GetBuffer();
-
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(serialized));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
 
     auto image_array = deserializer.GetDataMessage();
@@ -374,10 +412,8 @@ TEST_CASE("CBORSerialize_Image_Spots", "[CBOR]") {
 
     REQUIRE_NOTHROW(serializer.SerializeImage(message));
 
-    auto serialized = serializer.GetBuffer();
-
     JFJochFrameDeserializer deserializer;
-    REQUIRE_NOTHROW(deserializer.Process(serialized));
+    REQUIRE_NOTHROW(deserializer.Process(serializer.GetBuffer(), serializer.GetBufferSize()));
     REQUIRE(deserializer.GetType() == JFJochFrameDeserializer::Type::IMAGE);
 
     auto image_array = deserializer.GetDataMessage();
@@ -452,10 +488,9 @@ TEST_CASE("CBORSerialize_Start_stream2", "[CBOR]") {
 
     REQUIRE_NOTHROW(serializer.SerializeSequenceStart(message));
 
-    auto image = serializer.GetBuffer();
     stream2_msg *msg;
 
-    auto ret = stream2_parse_msg(image.data(), image.size(), &msg);
+    auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
     REQUIRE(ret == STREAM2_OK);
     CHECK(msg->type == STREAM2_MSG_START);
     auto msg2 = (stream2_start_msg *) msg;
@@ -499,10 +534,9 @@ TEST_CASE("CBORSerialize_End_stream2", "[CBOR]") {
 
     REQUIRE_NOTHROW(serializer.SerializeSequenceEnd(message));
 
-    auto image = serializer.GetBuffer();
     stream2_msg *msg;
 
-    auto ret = stream2_parse_msg(image.data(), image.size(), &msg);
+    auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
     REQUIRE(ret == STREAM2_OK);
     CHECK(msg->type == STREAM2_MSG_END);
     auto msg2 = (stream2_end_msg *) msg;
@@ -546,11 +580,10 @@ TEST_CASE("CBORSerialize_Image_compressed_stream2", "[CBOR]") {
     };
 
     REQUIRE_NOTHROW(serializer.SerializeImage(message));
-    auto cbor_image = serializer.GetBuffer();
 
     stream2_msg *msg;
 
-    auto ret = stream2_parse_msg(cbor_image.data(), cbor_image.size(), &msg);
+    auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
     REQUIRE(ret == STREAM2_OK);
     CHECK(msg->type == STREAM2_MSG_IMAGE);
     auto msg2 = (stream2_image_msg *) msg;
@@ -601,7 +634,7 @@ TEST_CASE("CBORSerialize_Image_uncompressed_stream2", "[CBOR]") {
 
     stream2_msg *msg;
 
-    auto ret = stream2_parse_msg(cbor_image.data(), cbor_image.size(), &msg);
+    auto ret = stream2_parse_msg(serializer.GetBuffer(), serializer.GetBufferSize(), &msg);
     REQUIRE(ret == STREAM2_OK);
     CHECK(msg->type == STREAM2_MSG_IMAGE);
     auto msg2 = (stream2_image_msg *) msg;

tests/FrameTransformationTest.cpp

@@ -39,14 +39,12 @@ TEST_CASE("FrameTransformation_Raw_NoCompression" ,"") {
 
     std::vector<int16_t> output(experiment.GetPixelsNum());
 
-    transformation.SetOutput(output.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
-
-    REQUIRE(transformation.PackStandardOutput() == experiment.GetPixelDepth() * experiment.GetPixelsNum());
+    transformation.Pack();
+    REQUIRE(transformation.SaveCompressedImage(output.data()) == experiment.GetPixelDepth() * experiment.GetPixelsNum());
 
     uint32_t diff_0 = 0;
     uint32_t diff_1 = 0;
@@ -81,14 +79,13 @@ TEST_CASE("FrameTransformation_Converted_NoCompression" ,"") {
 
     std::vector<int16_t> output(experiment.GetPixelsNum());
 
-    transformation.SetOutput(output.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
 
-    REQUIRE(transformation.PackStandardOutput() == experiment.GetPixelDepth() * experiment.GetPixelsNum());
+    transformation.Pack();
+    REQUIRE(transformation.SaveCompressedImage(output.data()) == experiment.GetPixelDepth() * experiment.GetPixelsNum());
 
     REQUIRE(input_0[511*1024]             == output[CONVERTED_MODULE_SIZE * (2 * nmodules - 2) + 0]);
     REQUIRE(input_0[511*1024+256]/2       == output[CONVERTED_MODULE_SIZE * (2 * nmodules - 2) + 258]);
@@ -134,15 +131,14 @@ TEST_CASE("FrameTransformation_Converted_bshuf_lz4" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
     output_compressed.resize(compressed_size);
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
@@ -195,16 +191,14 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-
-    transformation.SetOutput(output_compressed.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -257,15 +251,14 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_bin_2x2" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -305,16 +298,14 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_rle" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-
-    transformation.SetOutput(output_compressed.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -369,8 +360,6 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
@@ -392,7 +381,8 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation" ,"") {
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -440,8 +430,6 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_bin_2x2" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
     for (int frame = 0; frame < experiment.GetSummation(); frame++) {
         for (int i = 0; i < nmodules; i++) {
             REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
@@ -450,7 +438,8 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_bin_2x2" ,"") {
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -493,15 +482,14 @@ TEST_CASE("FrameTransformation_Converted_16bit_preview_summation" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
     REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data(), 0, 0));
     REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data(), 0, 0));
     REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data(), 0, 0));
     REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data(), 0, 0));
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -561,9 +549,7 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_2frames" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
-    for (int i = 0; i < nmodules; i++) {
+     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
@@ -574,7 +560,8 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_2frames" ,"") {
     }
 
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
@@ -596,7 +583,8 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_2frames" ,"") {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
 
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());
@@ -644,8 +632,6 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_negatives" ,"") {
 
     std::vector<char> output_compressed(experiment.GetMaxCompressedSize());
 
-    transformation.SetOutput(output_compressed.data());
-
     for (int i = 0; i < nmodules; i++) {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_0.data() + i * RAW_MODULE_SIZE, i, 0));
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
@@ -666,7 +652,8 @@ TEST_CASE("FrameTransformation_Converted_bshuf_zstd_summation_negatives" ,"") {
         REQUIRE_NOTHROW(transformation.ProcessModule(input_1.data() + i * RAW_MODULE_SIZE, i, 1));
     }
     size_t compressed_size;
-    REQUIRE_NOTHROW(compressed_size = transformation.PackStandardOutput());
+    transformation.Pack();
+    REQUIRE_NOTHROW(compressed_size = transformation.SaveCompressedImage(output_compressed.data()));
 
     REQUIRE(bshuf_read_uint64_BE(output_compressed.data()) == experiment.GetPixelsNum() * experiment.GetPixelDepth());
     REQUIRE(bshuf_read_uint32_BE(output_compressed.data()+8) == JFJochBitShuffleCompressor::DefaultBlockSize * experiment.GetPixelDepth());

tests/ZMQImagePusherTest.cpp

@@ -97,7 +97,7 @@ TEST_CASE("ZMQImageCommTest_1Writer","[ZeroMQ]") {
             data_message.number = i;
             PrepareCBORImage(data_message, x, image1.data() + i * x.GetPixelsNum(), x.GetPixelsNum() * sizeof(uint16_t));
             serializer.SerializeImage(data_message);
-            pusher.SendData(serializer.GetBuffer(), i);
+            pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
         }
 
         pusher.EndDataCollection(end_message);
@@ -172,7 +172,7 @@ TEST_CASE("ZMQImageCommTest_2Writers","[ZeroMQ]") {
             data_message.number = i;
             PrepareCBORImage(data_message, x, image1.data() + i * x.GetPixelsNum(), x.GetPixelsNum() * sizeof(uint16_t));
             serializer.SerializeImage(data_message);
-            pusher.SendData(serializer.GetBuffer(), i);
+            pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
         }
 
         pusher.EndDataCollection(end_message);
@@ -260,7 +260,7 @@ TEST_CASE("ZMQImageCommTest_4Writers","[ZeroMQ]") {
             data_message.number = i;
             PrepareCBORImage(data_message, x, image1.data() + i * x.GetPixelsNum(), x.GetPixelsNum() * sizeof(uint16_t));
             serializer.SerializeImage(data_message);
-            pusher.SendData(serializer.GetBuffer(), i);
+            pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
         }
 
         pusher.EndDataCollection(end_message);

tools/CompressionBenchmark.cpp

@@ -21,11 +21,11 @@ std::string CheckCompression(const DiffractionExperiment &x, size_t nimages, con
     auto start_time = std::chrono::system_clock::now();
 
     for (int i = 0; i < nimages; i++) {
-        transformation.SetOutput(output.data() + i * x.GetMaxCompressedSize());
         for (int j = 0; j < x.GetModulesNum(); j++ ) {
             transformation.ProcessModule(image.data() + (j + i * x.GetModulesNum()) * RAW_MODULE_SIZE, j, 0);
         }
-        compressed_size += transformation.PackStandardOutput();
+        transformation.Pack();
+        compressed_size += transformation.SaveCompressedImage(output.data() + i * x.GetMaxCompressedSize());
     }
 
     auto end_time = std::chrono::system_clock::now();
@@ -47,11 +47,11 @@ std::string CheckDecompression(const DiffractionExperiment &x, size_t nimages, c
     std::vector<size_t> compressed_size(nimages);
 
     for (int i = 0; i < nimages; i++) {
-        transformation.SetOutput(output[i].data());
         for (int j = 0; j < x.GetModulesNum(); j++ ) {
             transformation.ProcessModule(image.data() + (j + i * x.GetModulesNum()) * RAW_MODULE_SIZE, j, 0);
         }
-        compressed_size[i] = transformation.PackStandardOutput();
+        transformation.Pack();
+        compressed_size[i] = transformation.SaveCompressedImage(output[i].data());
         output[i].resize(compressed_size[i]);
     }
 

tools/HDF5DatasetWriteTest.cpp

@@ -87,10 +87,10 @@ int main(int argc, char **argv) {
     for (auto &i: output) i.resize(x.GetMaxCompressedSize());
 
     for (int i = 0; i < nimages; i++) {
-        transformation.SetOutput(output[i].data());
         for (int j = 0; j < 8; j++)
             transformation.ProcessModule(image + (i * x.GetModulesNum() + j) * RAW_MODULE_SIZE, j, 0);
-        output_size[i] = transformation.PackStandardOutput();
+        transformation.Pack();
+        output_size[i] = transformation.SaveCompressedImage(output[i].data());
     }
 
     x.ImagesPerTrigger(nimages_out);

writer/jfjoch_writer_test.cpp

@@ -79,10 +79,10 @@ int main(int argc, char **argv) {
                                image_tmp_raw.data(),
                                image_tmp_conv.data());
 
-        transformation.SetOutput(output[i].data());
         for (int j = 0; j < x.GetModulesNum(); j++)
             transformation.ProcessModule(image_tmp_raw.data() + j * RAW_MODULE_SIZE, j, 0);
-        output_size[i] = transformation.PackStandardOutput();
+        transformation.Pack();
+        output_size[i] = transformation.SaveCompressedImage((uint8_t *) output[i].data());
     }
 
     logger.Info("Sending {} images", nimages_out);
@@ -102,7 +102,7 @@ int main(int argc, char **argv) {
         data_message.number = i;
         PrepareCBORImage(data_message, x, output[i % nimages_in_file].data(), output_size[i % nimages_in_file]);
         serializer.SerializeImage(data_message);
-        pusher.SendData(serializer.GetBuffer(), i);
+        pusher.SendImage(serializer.GetBuffer(), serializer.GetBufferSize(), i);
     }
 
     EndMessage end_message{};