zmq2edi: detect FCT discontinuity

1 files changed, 68 insertions, 55 deletions

diff --git a/src/zmq2edi/zmq2edi.cpp b/src/zmq2edi/zmq2edi.cpp
index e69a6c2..20ddcbe 100644
--- a/src/zmq2edi/zmq2edi.cpp
+++ b/src/zmq2edi/zmq2edi.cpp
@@ -224,6 +224,8 @@ static void parse_destination_args(char option)
     }
 }
 
+class FCTDiscontinuity { };
+
 int start(int argc, char **argv)
 {
     edi_conf.enable_pft = true;
@@ -350,89 +352,100 @@ int start(int argc, char **argv)
     zmq::context_t zmq_ctx(1);
     etiLog.level(info) << "Opening ZMQ input: " << source_url;
 
-    size_t num_consecutive_resets = 0;
     while (true) {
         zmq::socket_t zmq_sock(zmq_ctx, ZMQ_SUB);
         zmq_sock.connect(source_url);
         zmq_sock.setsockopt(ZMQ_SUBSCRIBE, NULL, 0); // subscribe to all messages
 
         size_t error_count = 0;
+        int previous_fct = -1;
+
+        try {
+            while (error_count < MAX_ERROR_COUNT) {
+                zmq::message_t incoming;
+                zmq::pollitem_t items[1];
+                items[0].socket = zmq_sock;
+                items[0].events = ZMQ_POLLIN;
+                const int num_events = zmq::poll(items, 1, ZMQ_TIMEOUT_MS);
+                if (num_events == 0) { // timeout
+                    error_count++;
+                }
+                else {
+                    // Event received: recv will not block
+                    zmq_sock.recv(&incoming);
+                    const auto received_at = std::chrono::steady_clock::now();
 
-        while (error_count < MAX_ERROR_COUNT) {
-            zmq::message_t incoming;
-            zmq::pollitem_t items[1];
-            items[0].socket = zmq_sock;
-            items[0].events = ZMQ_POLLIN;
-            const int num_events = zmq::poll(items, 1, ZMQ_TIMEOUT_MS);
-            if (num_events == 0) { // timeout
-                error_count++;
-            }
-            else {
-                num_consecutive_resets = 0;
+                    zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
 
-                // Event received: recv will not block
-                zmq_sock.recv(&incoming);
-                const auto received_at = std::chrono::steady_clock::now();
+                    if (dab_msg->version != 1) {
+                        etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
+                        error_count++;
+                    }
 
-                zmq_dab_message_t* dab_msg = (zmq_dab_message_t*)incoming.data();
+                    int offset = sizeof(dab_msg->version) +
+                        NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
 
-                if (dab_msg->version != 1) {
-                    etiLog.level(error) << "ZeroMQ wrong packet version " << dab_msg->version;
-                    error_count++;
-                }
+                    std::vector<frame_t> all_frames;
+                    all_frames.reserve(NUM_FRAMES_PER_ZMQ_MESSAGE);
 
-                int offset = sizeof(dab_msg->version) +
-                    NUM_FRAMES_PER_ZMQ_MESSAGE * sizeof(*dab_msg->buflen);
+                    for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
+                        if (dab_msg->buflen[i] <= 0 or dab_msg->buflen[i] > 6144) {
+                            etiLog.level(error) << "ZeroMQ buffer " << i <<
+                                " has invalid length " << dab_msg->buflen[i];
+                            error_count++;
+                        }
+                        else {
+                            frame_t frame;
+                            frame.data.resize(6144, 0x55);
+                            frame.received_at = received_at;
 
-                std::vector<frame_t> all_frames;
-                all_frames.reserve(NUM_FRAMES_PER_ZMQ_MESSAGE);
+                            const int framesize = dab_msg->buflen[i];
 
-                for (int i = 0; i < NUM_FRAMES_PER_ZMQ_MESSAGE; i++) {
-                    if (dab_msg->buflen[i] <= 0 or dab_msg->buflen[i] > 6144) {
-                        etiLog.level(error) << "ZeroMQ buffer " << i <<
-                            " has invalid length " << dab_msg->buflen[i];
-                        error_count++;
-                    }
-                    else {
-                        frame_t frame;
-                        frame.data.resize(6144, 0x55);
-                        frame.received_at = received_at;
+                            memcpy(frame.data.data(),
+                                    ((uint8_t*)incoming.data()) + offset,
+                                    framesize);
 
-                        const int framesize = dab_msg->buflen[i];
+                            const int fct = frame.data[4];
 
-                        memcpy(frame.data.data(),
-                                ((uint8_t*)incoming.data()) + offset,
-                                framesize);
+                            const int expected_fct = (previous_fct + 1) % 250;
+                            if (previous_fct != -1 and expected_fct != fct) {
+                                etiLog.level(error) << "ETI wrong frame counter FCT=" << fct << " expected " << expected_fct;
+                                throw FCTDiscontinuity();
+                            }
+                            previous_fct = fct;
 
-                        all_frames.push_back(std::move(frame));
+                            all_frames.push_back(std::move(frame));
 
-                        offset += framesize;
+                            offset += framesize;
+                        }
                     }
-                }
 
-                for (auto &f : all_frames) {
-                    size_t consumed_bytes = 0;
+                    for (auto &f : all_frames) {
+                        size_t consumed_bytes = 0;
 
-                    f.metadata = get_md_one_frame(
-                            static_cast<uint8_t*>(incoming.data()) + offset,
-                            incoming.size() - offset,
-                            &consumed_bytes);
+                        f.metadata = get_md_one_frame(
+                                static_cast<uint8_t*>(incoming.data()) + offset,
+                                incoming.size() - offset,
+                                &consumed_bytes);
 
-                    offset += consumed_bytes;
-                }
+                        offset += consumed_bytes;
+                    }
 
-                for (auto &f : all_frames) {
-                    edisender.push_frame(std::move(f));
+                    for (auto &f : all_frames) {
+                        edisender.push_frame(std::move(f));
+                    }
                 }
             }
-        }
 
-        num_consecutive_resets++;
+            etiLog.level(info) << "Backoff " << backoff_after_reset_ms <<
+                "ms due to ZMQ input (" << source_url << ") timeout";
+        }
+        catch (const FCTDiscontinuity&) {
+            etiLog.level(info) << "Backoff " << backoff_after_reset_ms << "ms FCT discontinuity";
+        }
 
         zmq_sock.close();
         std::this_thread::sleep_for(std::chrono::milliseconds(backoff_after_reset_ms));
-        etiLog.level(info) << "ZMQ input (" << source_url << ") timeout after " <<
-            num_consecutive_resets << " consecutive resets.";
     }
 
     return 0;