Merge pull request #17 from Unit-X/star-1305-fix-demuxer-looses-sync

Star 1305 fix demuxer looses sync

Author: permobergedge

.github/workflows/unittests.yml

@@ -16,6 +16,7 @@ jobs:
     - name: get the ts test file
       run: |
         wget https://github.com/andersc/assets/raw/master/bars.ts
+        cp bars.ts bars-github.ts
     - name: build unit tests
       run: |
         mkdir build

mpegts/common.cpp

@@ -16,13 +16,13 @@ void writePts(SimpleBuffer &rSb, uint32_t lFb, uint64_t lPts) {
     uint32_t lVal;
 
     lVal = lFb << 4 | (((lPts >> 30) & 0x07) << 1) | 1;
-    rSb.write1Byte((int8_t) lVal);
+    rSb.write1Byte((uint8_t) lVal);
 
     lVal = (((lPts >> 15) & 0x7fff) << 1) | 1;
-    rSb.write2Bytes((int16_t) lVal);
+    rSb.write2Bytes((uint16_t) lVal);
 
     lVal = (((lPts) & 0x7fff) << 1) | 1;
-    rSb.write2Bytes((int16_t) lVal);
+    rSb.write2Bytes((uint16_t) lVal);
 }
 
 uint64_t readPts(SimpleBuffer &rSb) {

mpegts/mpegts_demuxer.cpp

@@ -18,12 +18,12 @@
 
 namespace mpegts {
 
-class MpegTsDemuxer {
+class MpegTsDemuxer final {
 public:
 
-    MpegTsDemuxer();
+    MpegTsDemuxer() = default;
 
-    virtual ~MpegTsDemuxer();
+    ~MpegTsDemuxer() = default;
 
     uint8_t decode(SimpleBuffer &rIn);
 
@@ -33,8 +33,9 @@ class MpegTsDemuxer {
     std::function<void(uint16_t pid, uint8_t expectedCC, uint8_t actualCC)> ccErrorCallback = nullptr;
 
     // stream, pid
+    const int kPatId = 0x0;
     std::map<uint8_t, int> mStreamPidMap;
-    int mPmtId;
+    int mPmtId = 0;
 
     // PAT
     PATHeader mPatHeader;
@@ -45,16 +46,46 @@ class MpegTsDemuxer {
     bool mPmtIsValid = false;
 
 private:
-    // pid, Elementary data frame
+    // Check that the CC is incremented according to spec. ccErrorCallback is called on error.
+    void checkContinuityCounter(const TsHeader &rTsHeader, uint8_t discontinuityIndicator);
+
+    // Check that the current position in buffer is pointing to a sync byte and if the buffer is larger than a TS packet,
+    // the function checks that the first byte after the TS packet is also a sync byte. Returns true if the buffer is in
+    // sync with the TS packet boundaries, otherwise false.
+    static bool checkSync(SimpleBuffer& rIn);
+
+    // Parse a TS packet, the input buffer should be aligned to the start of a TS packet and the buffer should be at least
+    // 188 bytes long.
+    void parseTsPacket(SimpleBuffer& rSb);
+
+    // Parse the adaptation field, will look for PCR values and pass to the pcrOutCallback. Returns true if parsing was
+    // successful, otherwise false. The randomAccessIndicator is set to 1 if the packet contains a random access point.
+    bool parseAdaptationField(const TsHeader& rTsHeader, SimpleBuffer& rSb, size_t bytesLeftInPacket, uint8_t& randomAccessIndicator);
+
+    // Parse the PAT table
+    bool parsePat(const TsHeader& rTsHeader, SimpleBuffer& rSb);
+
+    // Parse the PMT table
+    bool parsePmt(const TsHeader& rTsHeader, SimpleBuffer& rSb);
+
+    // Parse the PES data and create EsFrames. The EsFrames are passed to the esOutCallback.
+    bool parsePes(const TsHeader& rTsHeader, uint8_t randomAccessIndicator, SimpleBuffer& rSb, size_t payloadSize);
+
+    // Parse the PCR value from the adaptation field and pass to the pcrOutCallback.
+    void parsePcr(const AdaptationFieldHeader& rAdaptionField, SimpleBuffer& rSb) const;
+
+    // Check if the PCR value should be parsed based on the PID and the pcrOutCallback.
+    bool shouldParsePcr(uint16_t pid) const;
+
+    // Elementary stream data frames
     std::map<int, EsFrame> mEsFrames;
-    int mPcrId;
+    int mPcrId = 0;
     SimpleBuffer mRestData;
 
     // Continuity counters. Map from PID to current CC value.
     std::unordered_map<uint16_t, uint8_t> mCCs;
 
-    // Check that the CC is incremented according to spec. ccErrorCallback is called on error.
-    void checkContinuityCounter(const TsHeader &rTsHeader, uint8_t discontinuityIndicator);
+    size_t mBytesDroppedToRecoverSync = 0; // Keep track of how many bytes were dropped to recover sync
 };
 
 }

mpegts/mpegts_demuxer.h

@@ -1,66 +1,63 @@
 #include "simple_buffer.h"
-#include <assert.h>
+
+#include <cstring>
 #include <iterator>
 
+#include <cassert>
+
 namespace mpegts {
 
-SimpleBuffer::SimpleBuffer()
-    : mPos(0)
-{
-}
+SimpleBuffer::SimpleBuffer() = default;
 
 SimpleBuffer::SimpleBuffer(int32_t size, int8_t value)
-    : mPos(0)
 {
     mData = std::vector<uint8_t>(size, value);
 }
 
-SimpleBuffer::~SimpleBuffer()
-{
-}
+SimpleBuffer::~SimpleBuffer() = default;
 
-void SimpleBuffer::write1Byte(int8_t val)
+void SimpleBuffer::write1Byte(uint8_t val)
 {
     mData.push_back(val);
 }
 
-void SimpleBuffer::write2Bytes(int16_t val)
+void SimpleBuffer::write2Bytes(uint16_t val)
 {
-    char *p = (char *)&val;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 1; i >= 0; --i) {
         mData.push_back(p[i]);
     }
 }
 
-void SimpleBuffer::write3Bytes(int32_t val)
+void SimpleBuffer::write3Bytes(uint32_t val)
 {
-    char *p = (char *)&val;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 2; i >= 0; --i) {
         mData.push_back(p[i]);
     }
 }
 
-void SimpleBuffer::write4Bytes(int32_t val)
+void SimpleBuffer::write4Bytes(uint32_t val)
 {
-    char *p = (char *)&val;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 3; i >= 0; --i) {
         mData.push_back(p[i]);
     }
 }
 
-void SimpleBuffer::write8Bytes(int64_t val)
+void SimpleBuffer::write8Bytes(uint64_t val)
 {
-    char *p = (char *)&val;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 7; i >= 0; --i) {
         mData.push_back(p[i]);
     }
 }
 
-void SimpleBuffer::append(const uint8_t* bytes, int size)
+void SimpleBuffer::append(const uint8_t* bytes, size_t size)
 {
     if (!bytes || size <= 0) {
 #ifdef DEBUG
@@ -72,7 +69,7 @@ void SimpleBuffer::append(const uint8_t* bytes, int size)
     mData.insert(mData.end(), bytes, bytes + size);
 }
 
-void SimpleBuffer::prepend(const uint8_t* bytes, int size)
+void SimpleBuffer::prepend(const uint8_t* bytes, size_t size)
 {
     if (!bytes || size <= 0) {
 #ifdef DEBUG
@@ -84,22 +81,22 @@ void SimpleBuffer::prepend(const uint8_t* bytes, int size)
     mData.insert(mData.begin(), bytes, bytes + size);
 }
 
-int8_t SimpleBuffer::read1Byte()
+uint8_t SimpleBuffer::read1Byte()
 {
     assert(require(1));
 
-    int8_t val = mData.at(0 + mPos);
+    uint8_t val = mData.at(0 + mPos);
     mPos++;
 
     return val;
 }
 
-int16_t SimpleBuffer::read2Bytes()
+uint16_t SimpleBuffer::read2Bytes()
 {
     assert(require(2));
 
-    int16_t val = 0;
-    char *p = (char *)&val;
+    uint16_t val = 0;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 1; i >= 0; --i) {
         p[i] = mData.at(0 + mPos);
@@ -109,12 +106,12 @@ int16_t SimpleBuffer::read2Bytes()
     return val;
 }
 
-int32_t SimpleBuffer::read3Bytes()
+uint32_t SimpleBuffer::read3Bytes()
 {
     assert(require(3));
 
-    int32_t val = 0;
-    char *p = (char *)&val;
+    uint32_t val = 0;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 2; i >= 0; --i) {
         p[i] = mData.at(0 + mPos);
@@ -124,12 +121,12 @@ int32_t SimpleBuffer::read3Bytes()
     return val;
 }
 
-int32_t SimpleBuffer::read4Bytes()
+uint32_t SimpleBuffer::read4Bytes()
 {
     assert(require(4));
 
     int32_t val = 0;
-    char *p = (char *)&val;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 3; i >= 0; --i) {
         p[i] = mData.at(0 + mPos);
@@ -139,12 +136,12 @@ int32_t SimpleBuffer::read4Bytes()
     return val;
 }
 
-int64_t SimpleBuffer::read8Bytes()
+uint64_t SimpleBuffer::read8Bytes()
 {
     assert(require(8));
 
-    int64_t val = 0;
-    char *p = (char *)&val;
+    uint64_t val = 0;
+    uint8_t *p = reinterpret_cast<uint8_t*>(&val);
 
     for (int i = 7; i >= 0; --i) {
         p[i] = mData.at(0 + mPos);
@@ -154,46 +151,56 @@ int64_t SimpleBuffer::read8Bytes()
     return val;
 }
 
-std::string SimpleBuffer::readString(int len)
+std::string SimpleBuffer::readString(size_t len)
 {
     assert(require(len));
 
-    std::string val(*(char*)&mData[0] + mPos, len);
+    std::string val(reinterpret_cast<char*>(&mData.at(mPos)), len);
     mPos += len;
 
     return val;
 }
 
-void SimpleBuffer::skip(int size)
+void SimpleBuffer::skip(size_t size)
 {
     mPos += size;
 }
 
-bool SimpleBuffer::require(int required_size)
+bool SimpleBuffer::require(size_t required_size) const
 {
     assert(required_size >= 0);
 
     return required_size <= mData.size() - mPos;
 }
 
-bool SimpleBuffer::empty()
+bool SimpleBuffer::empty() const
 {
     return mPos >= mData.size();
 }
 
-int SimpleBuffer::size()
+size_t SimpleBuffer::size() const
 {
     return mData.size();
 }
 
-int SimpleBuffer::pos()
+size_t SimpleBuffer::pos() const
 {
     return mPos;
 }
 
 uint8_t* SimpleBuffer::data()
 {
-    return (size() == 0) ? nullptr : &mData[0];
+    return mData.empty() ? nullptr : &mData[0];
+}
+
+uint8_t* SimpleBuffer::currentData()
+{
+    return (mData.empty() || mPos >= mData.size()) ? nullptr : &mData[mPos];
+}
+
+size_t SimpleBuffer::dataLeft() const
+{
+    return mData.size() - mPos;
 }
 
 void SimpleBuffer::clear()
@@ -202,7 +209,7 @@ void SimpleBuffer::clear()
     mData.clear();
 }
 
-void SimpleBuffer::setData(int pos, const uint8_t* data, int len)
+void SimpleBuffer::setData(size_t pos, const uint8_t* data, size_t len)
 {
     if (!data) {
 #ifdef DEBUG
@@ -218,9 +225,7 @@ void SimpleBuffer::setData(int pos, const uint8_t* data, int len)
         return;
     }
 
-    for (int i = 0; i < len; i++) {
-        mData[pos + i] = data[i];
-    }
+    std::memcpy(currentData(), data, len);
 }
 
 }