ZMQStream2Pusher: Add error message for writer died during data collection

image_pusher/TCPStreamPusher.cpp

@@ -27,6 +27,7 @@ void TCPStreamPusher::StartDataCollection(StartMessage &message) {
     images_per_file = message.images_per_file;
     run_number = message.run_number;
     run_name = message.run_name;
+    transmission_error = false;
 
     for (size_t i = 0; i < socket.size(); i++) {
         if (!socket[i]->AcceptConnection(std::chrono::seconds(5)))
@@ -93,11 +94,14 @@ bool TCPStreamPusher::EndDataCollection(const EndMessage &message) {
         else if (!s->Send(serialization_buffer.data(), serializer.GetBufferSize(), TCPFrameType::END))
             ret = false;
     }
+    transmission_error = !ret;
     return ret;
 }
 
 std::string TCPStreamPusher::Finalize() {
     std::string ret;
+    if (transmission_error)
+        ret += "Timeout sending images (e.g., writer disabled during data collection);";
     if (writer_notification_socket) {
         for (size_t i = 0; i < socket.size(); i++) {
             auto n = writer_notification_socket->Receive(run_number, run_name);

image_pusher/TCPStreamPusher.h

@@ -16,7 +16,7 @@ class TCPStreamPusher : public ImagePusher {
     int64_t images_per_file = 1;
     uint64_t run_number = 0;
     std::string run_name;
-
+    std::atomic<bool> transmission_error = false;
 public:
     explicit TCPStreamPusher(const std::vector<std::string>& addr,
                               std::optional<int32_t> send_buffer_size = {},

image_pusher/ZMQStream2Pusher.cpp

@@ -43,6 +43,7 @@ void ZMQStream2Pusher::StartDataCollection(StartMessage& message) {
     images_per_file = message.images_per_file;
     run_number = message.run_number;
     run_name = message.run_name;
+    transmission_error = false;
 
     for (int i = 0; i < socket.size(); i++) {
         message.socket_number = i;
@@ -77,6 +78,7 @@ bool ZMQStream2Pusher::EndDataCollection(const EndMessage& message) {
         if (!s->Send(serialization_buffer.data(), serializer.GetBufferSize()))
             ret = false;
     }
+    transmission_error = !ret;
     return ret;
 }
 
@@ -89,6 +91,8 @@ std::vector<std::string> ZMQStream2Pusher::GetAddress() {
 
 std::string ZMQStream2Pusher::Finalize() {
     std::string ret;
+    if (transmission_error)
+        ret += "Timeout sending images (e.g., writer disabled during data collection);";
     if (writer_notification_socket) {
         for (int i = 0; i < socket.size(); i++) {
             auto n = writer_notification_socket->Receive(run_number, run_name);

image_pusher/ZMQStream2Pusher.h

@@ -21,6 +21,7 @@ class ZMQStream2Pusher : public ImagePusher {
     int64_t images_per_file = 1;
     uint64_t run_number = 0;
     std::string run_name;
+    std::atomic<bool> transmission_error = false;
 public:
     explicit ZMQStream2Pusher(const std::vector<std::string>& addr,
                               std::optional<int32_t> send_buffer_high_watermark = {},