Add metadata latency for all PipelinedModCodec

5 files changed, 72 insertions, 33 deletions

diff --git a/src/DabMod.cpp b/src/DabMod.cpp
index 25a93bf..4a4ea82 100644
--- a/src/DabMod.cpp
+++ b/src/DabMod.cpp
@@ -426,6 +426,12 @@ int launch_modulator(int argc, char* argv[])
             }
 #endif
 
+            // TODO remove
+            auto output_as_file = dynamic_pointer_cast<OutputFile>(output);
+            if (output_as_file) {
+                output_as_file->setETISource(modulator->getEtiSource());
+            }
+
             inputReader->PrintInfo();
 
             run_modulator_state_t st = run_modulator(m);
diff --git a/src/ModPlugin.cpp b/src/ModPlugin.cpp
index c39d883..f86bfb2 100644
--- a/src/ModPlugin.cpp
+++ b/src/ModPlugin.cpp
@@ -74,9 +74,9 @@ int ModOutput::process(
 
 PipelinedModCodec::PipelinedModCodec() :
     ModCodec(),
-    m_number_of_runs(0),
     m_input_queue(),
     m_output_queue(),
+    m_metadata_fifo(),
     m_running(false),
     m_thread()
 {
@@ -107,7 +107,7 @@ int PipelinedModCodec::process(Buffer* dataIn, Buffer* dataOut)
 
     m_input_queue.push(inbuffer);
 
-    if (m_number_of_runs > 0) {
+    if (m_ready_to_output_data) {
         std::shared_ptr<Buffer> outbuffer;
         m_output_queue.wait_and_pop(outbuffer);
 
@@ -116,13 +116,26 @@ int PipelinedModCodec::process(Buffer* dataIn, Buffer* dataOut)
     else {
         dataOut->setLength(dataIn->getLength());
         memset(dataOut->getData(), 0, dataOut->getLength());
-        m_number_of_runs++;
+        m_ready_to_output_data = true;
     }
 
     return dataOut->getLength();
 
 }
 
+meta_vec_t PipelinedModCodec::process_metadata(const meta_vec_t& metadataIn)
+{
+    m_metadata_fifo.push_back(metadataIn);
+    if (m_metadata_fifo.size() == 2) {
+        auto r = std::move(m_metadata_fifo.front());
+        m_metadata_fifo.pop_front();
+        return r;
+    }
+    else {
+        return {};
+    }
+}
+
 void PipelinedModCodec::process_thread()
 {
     set_thread_name(name());
diff --git a/src/ModPlugin.h b/src/ModPlugin.h
index 3c3e8b3..c0f1c1a 100644
--- a/src/ModPlugin.h
+++ b/src/ModPlugin.h
@@ -30,16 +30,34 @@
 #   include <config.h>
 #endif
 
-
 #include "Buffer.h"
 #include "ThreadsafeQueue.h"
-
-#include <sys/types.h>
+#include <cstddef>
 #include <vector>
 #include <memory>
 #include <thread>
 #include <atomic>
 
+// All flowgraph elements derive from ModPlugin, or a variant of it.
+// Some ModPlugins also support handling metadata.
+
+struct frame_timestamp;
+struct flowgraph_metadata {
+    std::shared_ptr<struct frame_timestamp> ts;
+};
+
+using meta_vec_t = std::vector<flowgraph_metadata>;
+
+/* ModPlugins that support metadata derive from ModMetadata */
+class ModMetadata {
+    public:
+        // Receives metadata from all inputs, and process them, and output
+        // a sequence of metadata.
+        virtual meta_vec_t process_metadata(const meta_vec_t& metadataIn) = 0;
+};
+
+
+/* Abstract base class for all flowgraph elements */
 class ModPlugin
 {
 public:
@@ -69,7 +87,11 @@ public:
     virtual int process(Buffer* const dataIn, Buffer* dataOut) = 0;
 };
 
-class PipelinedModCodec : public ModCodec
+/* Pipelined ModCodecs run their processing in a separate thread, and
+ * have a one-call-to-process() latency. Because of this latency, they
+ * must also handle the metadata
+ */
+class PipelinedModCodec : public ModCodec, public ModMetadata
 {
 public:
     PipelinedModCodec();
@@ -82,6 +104,8 @@ public:
     virtual int process(Buffer* const dataIn, Buffer* dataOut) final;
     virtual const char* name() = 0;
 
+    virtual meta_vec_t process_metadata(const meta_vec_t& metadataIn) final;
+
 protected:
     // Once the instance implementing PipelinedModCodec has been constructed,
     // it must call start_pipeline_thread()
@@ -89,11 +113,13 @@ protected:
     virtual int internal_process(Buffer* const dataIn, Buffer* dataOut) = 0;
 
 private:
-    size_t m_number_of_runs;
+    bool m_ready_to_output_data = false;
 
     ThreadsafeQueue<std::shared_ptr<Buffer> > m_input_queue;
     ThreadsafeQueue<std::shared_ptr<Buffer> > m_output_queue;
 
+    std::deque<meta_vec_t> m_metadata_fifo;
+
     std::atomic<bool> m_running;
     std::thread m_thread;
     void process_thread(void);
@@ -120,19 +146,3 @@ public:
     virtual int process(Buffer* dataIn) = 0;
 };
 
-struct frame_timestamp;
-struct flowgraph_metadata {
-    std::shared_ptr<struct frame_timestamp> ts;
-};
-
-
-using meta_vec_t = std::vector<flowgraph_metadata>;
-
-/* Some ModPlugins also support metadata */
-class ModMetadata {
-    public:
-        // Receives metadata from all inputs, and process them, and output
-        // a sequence of metadata.
-        virtual meta_vec_t process_metadata(const meta_vec_t& metadataIn) = 0;
-};
-
diff --git a/src/OutputFile.cpp b/src/OutputFile.cpp
index 3bb45c4..481e858 100644
--- a/src/OutputFile.cpp
+++ b/src/OutputFile.cpp
@@ -77,22 +77,26 @@ meta_vec_t OutputFile::process_metadata(const meta_vec_t& metadataIn)
 
     for (const auto& md : metadataIn) {
         if (md.ts) {
-            ss << "FCT=" << md.ts->fct <<
+            ss << " FCT=" << md.ts->fct <<
                 " FP=" << (int)md.ts->fp;
-            if (md.ts->timestamp_valid) {
-                ss << " ts=" << md.ts->timestamp_sec <<
-                    "+" << md.ts->timestamp_pps << ", ";
-            }
-            else {
-                ss << " no ts";
-            }
         }
         else {
-            ss << "void, ";
+            ss << " void, ";
         }
     }
 
+    if (myEtiSource) {
+        frame_timestamp ts;
+        myEtiSource->calculateTimestamp(ts);
+        ss << " ETI FCT=" << ts.fct;
+    }
+
     etiLog.level(debug) << "Output File got metadata: " << ss.str();
 
     return {};
 }
+
+void OutputFile::setETISource(EtiSource *etiSource)
+{
+    myEtiSource = etiSource;
+}
diff --git a/src/OutputFile.h b/src/OutputFile.h
index 2b92369..97fdcb7 100644
--- a/src/OutputFile.h
+++ b/src/OutputFile.h
@@ -32,6 +32,7 @@
 
 
 #include "ModPlugin.h"
+#include "EtiReader.h"
 
 #include <string>
 #include <stdio.h>
@@ -57,7 +58,12 @@ public:
     virtual meta_vec_t process_metadata(
             const meta_vec_t& metadataIn) override;
 
+    void setETISource(EtiSource *etiSource);
+
 protected:
+    // TODO remove
+    EtiSource *myEtiSource = nullptr;
+
     std::string myFilename;
     std::unique_ptr<FILE, FILEDeleter> myFile;
 };