ZMQStream2PusherSocket: Minor change to avert (imaginary) race condition

image_pusher/ZMQStream2PusherSocket.cpp

@@ -31,9 +31,8 @@ bool ZMQStream2PusherSocket::Send(const uint8_t *data, size_t size) {
 
 void ZMQStream2PusherSocket::StartWriterThread() {
     std::unique_lock ul(m);
-
-    send_future = std::async(std::launch::async, &ZMQStream2PusherSocket::WriterThread, this);
     active = true;
+    send_future = std::async(std::launch::async, &ZMQStream2PusherSocket::WriterThread, this);
 }
 
 void ZMQStream2PusherSocket::StopWriterThread() {
@@ -56,8 +55,10 @@ void ZMQStream2PusherSocket::WriterThread() {
     while (!e.end) {
         // When hitting timeout on SendZeroCopy consider it a transmission error and switch to non-blocking communication
         bool blocking_send = !transmission_error;
-        if (!s.SendZeroCopy(e.z->GetImage(), e.z->GetImageSize(), zmq_socket_free, e.z, blocking_send))
+        if (!s.SendZeroCopy(e.z->GetImage(), e.z->GetImageSize(), zmq_socket_free, e.z, blocking_send)) {
             transmission_error = true;
+            e.z->release(); // Important: callback won't run on failed enqueue/send
+        }
         e = queue.GetBlocking();
     }
 }

image_pusher/ZMQStream2PusherSocket.h

@@ -16,7 +16,7 @@
 
 class ZMQStream2PusherSocket {
     std::mutex m;
-    std::atomic<bool> active = false;
+    bool active = false;
     std::future<void> send_future;
 
     ThreadSafeFIFO<ImagePusherQueueElement> queue;