Refactor how frames are sent from receiver

frame_serialize/ZMQStream2Pusher.cpp

@@ -25,14 +25,11 @@ void ZMQStream2Pusher::Bind(const std::string &addr, int32_t send_buffer_high_wa
     socket.Bind(addr);
 }
 
-void ZMQStream2Pusher::StartDataCollection(const StartMessage &message) {
+void ZMQStream2Pusher::StartDataCollection(StartMessage &message) {
     size_t approx_size = 1024*1024;
     for (const auto &x : message.pixel_mask)
         approx_size += x.size;
 
-    for (const auto &x : message.calibration)
-        approx_size += x.size;
-
     std::vector<uint8_t> serialization_buffer(approx_size);
     CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
     serializer.SerializeSequenceStart(message);
@@ -40,14 +37,13 @@ void ZMQStream2Pusher::StartDataCollection(const StartMessage &message) {
         throw JFJochException(JFJochExceptionCategory::ZeroMQ, "Timeout on pushing start message on addr " + GetAddress());
 }
 
-bool ZMQStream2Pusher::SendImage(const DataMessage &message) {
-    size_t approx_size = message.image.size + 2*1024*1024;
-    std::vector<uint8_t> serialization_buffer(approx_size);
+bool ZMQStream2Pusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number) {
+    return socket.Send(image_data, image_size, false);
+}
 
-    CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
-    serializer.SerializeImage(message);
-
-    return socket.Send(serialization_buffer.data(), serializer.GetBufferSize(), false); // Non-blocking
+void ZMQStream2Pusher::SendImage(const uint8_t *image_data, size_t image_size, int64_t image_number,
+                                 ZeroCopyReturnValue *z) {
+    socket.SendZeroCopy(image_data,image_size, z);
 }
 
 bool ZMQStream2Pusher::EndDataCollection(const EndMessage &message) {
@@ -58,6 +54,13 @@ bool ZMQStream2Pusher::EndDataCollection(const EndMessage &message) {
     return socket.Send(serialization_buffer.data(), serializer.GetBufferSize(), true); // Blocking
 }
 
+bool ZMQStream2Pusher::SendCalibration(const CompressedImage &message) {
+    std::vector<uint8_t> serialization_buffer(80 * 1024 * 1024);
+    CBORStream2Serializer serializer(serialization_buffer.data(), serialization_buffer.size());
+    serializer.SerializeCalibration(message);
+    return socket.Send(serialization_buffer.data(), serializer.GetBufferSize(), true); // Blocking
+}
+
 std::string ZMQStream2Pusher::GetAddress() {
     return socket.GetEndpointName();
-}
+}