Add InputBuffered

5 files changed, 220 insertions, 53 deletions

diff --git a/src/inputs/Input.h b/src/inputs/Input.h
index 42c2ef7..a6a7231 100644
--- a/src/inputs/Input.h
+++ b/src/inputs/Input.h
@@ -31,6 +31,7 @@
 #endif
 
 #include <vector>
+#include <list>
 #include <string>
 #include <stdint.h>
 
@@ -38,41 +39,53 @@ class InputBase {
 
     public:
         virtual int Open();
-        virtual int Lock(); // TODO get rid of lock/unlock
-        virtual int Unlock();
+
+        // Must return a string with a name, used for logging
+        virtual const char* GetName();
 
         // ReadFrame must return exactly the right number of bytes
         // for the format, which means it must fill the buffer to
         // <size> bytes.
         virtual int ReadFrame(void* buffer, int size);
 
-        // TODO: ehm, where do we need this ?
-        virtual int SetBitrate(int bitrate);
-
         virtual int Close();
         virtual int Clean(); // TODO destructor
 };
 
-/********************************************/
-/**   TEST INPUT      ***********************/
-/********************************************/
+enum BufferState
+{
+    PREBUFFERING = 0,
+    RUNNING
+};
 
-// Implement mostly everything, but don't do anything
-class InputTest : InputBase {
+class InputBuffered : public InputBase {
     public:
-        InputTest() : counter(0) {}
-        int Open();
-        int Lock();
-        int Unlock();
-        int ReadFrame(void* buffer, int size);
-        int SetBitrate(int bitrate);
-        int Close();
-        int Clean();
+        virtual int ReadFrame(void* buffer, int size);
+    protected:
+        InputBuffered(int prebuffer_stages, int source_size,
+                int overfull_thresh) :
+                m_source_size(source_size),
+                m_prebuffer_stages(prebuffer_stages),
+                m_overfull_thresh(overfull_thresh),
+                m_bufferstate(PREBUFFERING)
+                 { } ;
+
+        // Reads a frame from the source, into the buffer
+        virtual int ReadSource(void* buffer, int size);
+
+        virtual bool FillBuffer();
 
     private:
-        unsigned long counter;
-};
+        InputBuffered() {};
 
+        int m_source_size;
+        int m_prebuffer_count;
+        int m_prebuffer_stages;
+        int m_overfull_thresh;
+        BufferState m_bufferstate;
+
+        std::list< std::vector<char> > m_buffer;
+};
 
 /********************************************/
 /**   FILE INPUTS     ***********************/
@@ -84,39 +97,73 @@ class InputTest : InputBase {
 // for Fifo, and also for different file types.
 class InputFile : InputBase {
     public:
-        InputFile(std::string fileName) : filename(fileName) {}
-        int Open();
-        int Close();
-        int Rewind();
+        InputFile(std::string fileName) : m_filename(fileName) {}
+        virtual int Open();
+        virtual int Close();
+        virtual int Rewind();
     protected:
-        long ReadData(void* data, size_t size, unsigned int tries);
-        std::string filename;
+        std::string m_filename;
         int file; // the file descriptor
 };
 
 #ifdef HAVE_FORMAT_MPEG
-class InputMpegFile : InputFile {
+class InputMpegFile : public InputFile, public InputBuffered {
     public:
-        InputMpegFile(std::string fileName) : filename(fileName) {}
+        InputMpegFile(std::string fileName) :
+            InputFile(fileName),
+            InputBuffered(prebuffer_stages, source_size, overfull_thresh)
+            { m_name = "mpeg " + m_filename;}
         int ReadFrame(void* buffer, int size);
+        const char* GetName();
     private:
+        std::string m_name;
+        int ReadSource(void* buffer, int size);
         bool parity;
 };
 #endif // HAVE_FORMAT_MPEG
 
 #ifdef HAVE_FORMAT_DABPLUS
-class InputDabplusFile : InputFile {
+class InputDabplusFile : public InputFile, public InputBuffered {
     public:
-        InputMpegFile(std::string fileName) : filename(fileName) {}
+        InputDabplusFile(
+                std::string fileName,
+                int prebuffer_stages,
+                int frame_size,
+                int overfull_thresh) :
+            InputFile(fileName),
+            InputBuffered(prebuffer_stages, frame_size, overfull_thresh)
+            { m_name = "dabplus " + m_filename;}
         int ReadFrame(void* buffer, int size);
+        const char* GetName();
     private:
-        int Read(void* buffer, int size);
+        std::string m_name;
+        int ReadSuperFrame(void* buffer, int size);
+        int ReadSource(void* buffer, int size);
         std::vector<uint8_t> buffer;
         size_t bufferIndex;
 };
 #endif
 
-#endif
+#endif // INPUT_FILE
+
+
+/********************************************/
+/**   TEST INPUT      ***********************/
+/********************************************/
+
+// Implement mostly everything, but don't do anything
+class InputTest : InputBase {
+    public:
+        InputTest() : counter(0) {}
+        int Open();
+        int ReadFrame(void* buffer, int size);
+        const char* GetName();
+        int Close();
+        int Clean();
+
+    private:
+        unsigned long counter;
+};
 
 #endif
 
diff --git a/src/inputs/InputBuffered.cpp b/src/inputs/InputBuffered.cpp
new file mode 100644
index 0000000..9eea2a7
--- /dev/null
+++ b/src/inputs/InputBuffered.cpp
@@ -0,0 +1,131 @@
+/*
+   Copyright (C) 2009 Her Majesty the Queen in Right of Canada (Communications
+   Research Center Canada)
+
+   Copyright (C) 2013 Matthias P. Braendli
+   http://mpb.li
+
+   Base class for buffered inputs
+
+   */
+/*
+   This file is part of CRC-DabMux.
+
+   CRC-DabMux is free software: you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as
+   published by the Free Software Foundation, either version 3 of the
+   License, or (at your option) any later version.
+
+   CRC-DabMux is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with CRC-DabMux.  If not, see <http://www.gnu.org/licenses/>.
+   */
+
+#include "inputs/Input.h"
+
+
+#include "TcpLog.h"
+
+#include <string>
+#include <cstring>
+#include <cstdlib>
+#include <cstdio>
+#include <fcntl.h>
+#include <unistd.h>
+#include <errno.h>
+
+#ifndef _WIN32
+#   define O_BINARY 0
+#endif
+
+extern TcpLog etiLog;
+
+int InputBuffered::ReadFrame(void* buffer, int size)
+{
+    int retval = 0;
+    bool rc;
+
+    switch (m_bufferstate) {
+        case PREBUFFERING:
+            rc = FillBuffer();
+            if (rc) {
+                m_prebuffer_count--;
+            }
+            if (m_prebuffer_count == 0) {
+                etiLog.print(TcpLog::NOTICE, "input %s: pre-buffering complete\n",
+                        GetName());
+                m_bufferstate = RUNNING;
+            }
+            break;
+
+        case RUNNING:
+            rc = FillBuffer();
+            if (m_buffer.empty()) {
+                etiLog.print(TcpLog::WARNING,
+                        "input %s input empty, re-enabling pre-buffering\n",
+                        GetName());
+
+                // reinitialise prebuffering
+                m_bufferstate = PREBUFFERING;
+                m_prebuffer_count = m_prebuffer_stages;
+            }
+            else {
+                std::vector<char> &buffer_head = m_buffer.front();
+
+                memcpy(buffer, &buffer_head.front(), size);
+                etiLog.print(TcpLog::NOTICE,
+                        "input %s got %d, %d\n",
+                        GetName(), rc, m_buffer.size());
+
+                m_buffer.pop_front();
+                retval = size;
+            }
+            break;
+    }
+
+    // If we haven't filled buffer, return a nulled one
+    if (retval == 0) {
+        memset(buffer, 0, size);
+        retval = size;
+    }
+
+    return retval;
+}
+
+// This implemenation of FillBuffer assumes that
+// input and output frames have same size.
+// That's not the case with DAB+ superframes
+bool InputBuffered::FillBuffer()
+{
+    if (m_buffer.size() > m_overfull_thresh) {
+        etiLog.print(TcpLog::WARNING,
+                "input %s buffer full, dropping frame !\n",
+                GetName());
+
+        // drop the surplus
+        int num_frames_to_drop = m_buffer.size() - m_overfull_thresh;
+        while (num_frames_to_drop--)
+            m_buffer.pop_front();
+
+        return true;
+    }
+    else {
+        // copy the input frame into the frame_buffer one to one
+        std::vector<char> frame(m_source_size);
+        int rc = ReadSource(&frame.front(), m_source_size);
+        if (rc == m_source_size)
+        {
+            m_buffer.push_back(frame);
+            return true;
+        }
+        else
+        {
+            return false;
+        }
+    }
+}
+
diff --git a/src/inputs/InputDabplusFile.cpp b/src/inputs/InputDabplusFile.cpp
index b55b3d9..ad76f9c 100644
--- a/src/inputs/InputDabplusFile.cpp
+++ b/src/inputs/InputDabplusFile.cpp
@@ -49,7 +49,7 @@
 
 extern TcpLog etiLog;
 
-int InputDabplusFile::Read(void* buffer, int size)
+int InputDabplusFile::ReadSuperFrame(void* buffer, int size)
 {
     if (this->buffer.size() != size * 5) {
         this->buffer.resize(size * 5);
@@ -73,12 +73,12 @@ int InputDabplusFile::Read(void* buffer, int size)
     return size;
 }
 
-int InputDabplusFile::ReadFrame(void* buffer, int size)
+int InputDabplusFile::ReadSource(void* buffer, int size)
 {
     int result;
     uint8_t* dataOut = reinterpret_cast<uint8_t*>(buffer);
 
-    result = Read(dataOut, size);
+    result = ReadSuperFrame(dataOut, size);
     if (result == -1) {
         etiLog.print(TcpLog::CRIT, "ERROR: Can't read file\n");
         perror("");
@@ -92,7 +92,7 @@ int InputDabplusFile::ReadFrame(void* buffer, int size)
             return -1;
         }
 
-        result = Read(dataOut + sizeOut, size - sizeOut);
+        result = ReadSuperFrame(dataOut + sizeOut, size - sizeOut);
         if (result == -1) {
             etiLog.print(TcpLog::CRIT, "ERROR: Can't read file\n");
             perror("");
diff --git a/src/inputs/InputMpegFile.cpp b/src/inputs/InputMpegFile.cpp
index 265b0ee..480a251 100644
--- a/src/inputs/InputMpegFile.cpp
+++ b/src/inputs/InputMpegFile.cpp
@@ -53,6 +53,11 @@ extern TcpLog etiLog;
 #define MPEG_ORIGINAL       -5
 #define MPEG_EMPHASIS       -6
 
+const char* InputMpegFile::GetName()
+{
+    return m_name.c_str();
+}
+
 int checkDabMpegFrame(void* data)
 {
     mpegHeader* header = (mpegHeader*)data;
@@ -79,7 +84,7 @@ int checkDabMpegFrame(void* data)
     return -1;
 }
 
-int InputMpegFile::ReadFrame(void* buffer, int size)
+int InputMpegFile::ReadSource(void* buffer, int size)
 {
     int result;
     bool rewind = false;
diff --git a/src/inputs/InputTest.cpp b/src/inputs/InputTest.cpp
index 4d2881e..6510bd1 100644
--- a/src/inputs/InputTest.cpp
+++ b/src/inputs/InputTest.cpp
@@ -40,16 +40,6 @@ int InputTest::Open()
     return 0;
 }
 
-int InputTest::Lock()
-{
-    return 0;
-}
-
-int InputTest::Unlock()
-{
-    return 0;
-}
-
 int InputTest::ReadFrame(void* buffer, int size)
 {
     char* data = (char*)buffer;
@@ -62,12 +52,6 @@ int InputTest::ReadFrame(void* buffer, int size)
     return size;
 }
 
-int InputTest::SetBitrate(int bitrate)
-{
-    return bitrate;
-}
-
-
 int InputTest::Close()
 {
     return 0;