for #250, decode the PAT of PSI ts packet.

README.md

@@ -486,6 +486,8 @@ Supported operating systems and hardware:
 ).
 1. Support HLS(h.264+mp3) streaming, read 
 [#301](https://github.com/winlinvip/simple-rtmp-server/issues/301).
+1. [dev] Support push MPEG-TS over UDP to SRS, read 
+[#250](https://github.com/winlinvip/simple-rtmp-server/issues/250).
 1. [no-plan] Support <500ms latency, FRSC(Fast RTMP-compatible Stream Channel tech).
 1. [no-plan] Support RTMP 302 redirect [#92](https://github.com/winlinvip/simple-rtmp-server/issues/92).
 1. [no-plan] Support multiple processes, for both origin and edge

trunk/src/app/srs_app_mpegts_udp.cpp

@@ -35,19 +35,20 @@ using namespace std;
 #include <srs_kernel_ts.hpp>
 #include <srs_kernel_stream.hpp>
 #include <srs_kernel_ts.hpp>
-#include <srs_core_autofree.hpp>
 
 #ifdef SRS_AUTO_STREAM_CASTER
 
 SrsMpegtsOverUdp::SrsMpegtsOverUdp(SrsConfDirective* c)
 {
     stream = new SrsStream();
+    context = new SrsTsContext();
     output = _srs_config->get_stream_caster_output(c);
 }
 
 SrsMpegtsOverUdp::~SrsMpegtsOverUdp()
 {
     srs_freep(stream);
+    srs_freep(context);
 }
 
 int SrsMpegtsOverUdp::on_udp_packet(sockaddr_in* from, char* buf, int nb_buf)
@@ -85,10 +86,7 @@ int SrsMpegtsOverUdp::on_ts_packet(SrsStream* stream)
 {
     int ret = ERROR_SUCCESS;
 
-    SrsTsPacket* packet = new SrsTsPacket();
+    if ((ret = context->decode(stream)) != ERROR_SUCCESS) {
-    SrsAutoFree(SrsTsPacket, packet);
-
-    if ((ret = packet->decode(stream)) != ERROR_SUCCESS) {
         srs_error("mpegts: decode ts packet failed. ret=%d", ret);
         return ret;
     }

trunk/src/app/srs_app_mpegts_udp.hpp

@@ -34,6 +34,7 @@ class sockaddr_in;
 #include <string>
 
 class SrsStream;
+class SrsTsContext;
 class SrsConfDirective;
 
 #ifdef SRS_AUTO_STREAM_CASTER
@@ -45,6 +46,7 @@ class SrsMpegtsOverUdp
 {
 private:
     SrsStream* stream;
+    SrsTsContext* context;
     std::string output;
 public:
     SrsMpegtsOverUdp(SrsConfDirective* c);

trunk/src/kernel/srs_kernel_ts.cpp

@@ -39,6 +39,7 @@ using namespace std;
 #include <srs_kernel_buffer.hpp>
 #include <srs_kernel_utility.hpp>
 #include <srs_kernel_stream.hpp>
+#include <srs_core_autofree.hpp>
 
 // in ms, for HLS aac sync time.
 #define SRS_CONF_DEFAULT_AAC_SYNC 100
@@ -401,6 +402,33 @@ SrsMpegtsFrame::SrsMpegtsFrame()
     key = false;
 }
 
+SrsTsContext::SrsTsContext()
+{
+}
+
+SrsTsContext::~SrsTsContext()
+{
+}
+
+int SrsTsContext::decode(SrsStream* stream)
+{
+    int ret = ERROR_SUCCESS;
+
+    // parse util EOF of stream.
+    // for example, parse multiple times for the PES_packet_length(0) packet.
+    while (!stream->empty()) {
+        SrsTsPacket* packet = new SrsTsPacket();
+        SrsAutoFree(SrsTsPacket, packet);
+
+        if ((ret = packet->decode(stream)) != ERROR_SUCCESS) {
+            srs_error("mpegts: decode ts packet failed. ret=%d", ret);
+            return ret;
+        }
+    }
+
+    return ret;
+}
+
 SrsTsPacket::SrsTsPacket()
 {
     sync_byte = 0;
@@ -412,11 +440,13 @@ SrsTsPacket::SrsTsPacket()
     adaption_field_control = SrsTsAdaptationFieldTypeReserved;
     continuity_counter = 0;
     adaptation_field = NULL;
+    payload = NULL;
 }
 
 SrsTsPacket::~SrsTsPacket()
 {
     srs_freep(adaptation_field);
+    srs_freep(payload);
 }
 
 int SrsTsPacket::decode(SrsStream* stream)
@@ -471,6 +501,23 @@ int SrsTsPacket::decode(SrsStream* stream)
     // calc the user defined data size for payload.
     int nb_payload = SRS_TS_PACKET_SIZE - (stream->pos() - pos);
 
+    // optional: payload.
+    if (adaption_field_control == SrsTsAdaptationFieldTypePayloadOnly || adaption_field_control == SrsTsAdaptationFieldTypeBoth) {
+        if (pid == SrsTsPidPAT) {
+            // 2.4.4.3 Program association Table
+            srs_freep(payload);
+            payload = new SrsTsPayloadPAT(this);
+        } else {
+            // left bytes as reserved.
+            stream->skip(nb_payload);
+        }
+
+        if (payload && (ret = payload->decode(stream)) != ERROR_SUCCESS) {
+            srs_error("ts: demux payload failed. ret=%d", ret);
+            return ret;
+        }
+    }
+
     return ret;
 }
 
@@ -713,6 +760,145 @@ int SrsTsAdaptationField::decode(SrsStream* stream)
     return ret;
 }
 
+SrsTsPayloadPATProgram::SrsTsPayloadPATProgram()
+{
+    number = 0;
+    pid = 0;
+}
+
+SrsTsPayloadPATProgram::~SrsTsPayloadPATProgram()
+{
+}
+
+SrsTsPayload::SrsTsPayload(SrsTsPacket* p)
+{
+    packet = p;
+}
+
+SrsTsPayload::~SrsTsPayload()
+{
+}
+
+SrsTsPayloadPSI::SrsTsPayloadPSI(SrsTsPacket* p) : SrsTsPayload(p)
+{
+    pointer_field = 0;
+}
+
+SrsTsPayloadPSI::~SrsTsPayloadPSI()
+{
+}
+
+int SrsTsPayloadPSI::decode(SrsStream* stream)
+{
+    int ret = ERROR_SUCCESS;
+
+    /**
+    * When the payload of the Transport Stream packet contains PSI data, the payload_unit_start_indicator has the following
+    * significance: if the Transport Stream packet carries the first byte of a PSI section, the payload_unit_start_indicator value
+    * shall be '1', indicating that the first byte of the payload of this Transport Stream packet carries the pointer_field. If the
+    * Transport Stream packet does not carry the first byte of a PSI section, the payload_unit_start_indicator value shall be '0',
+    * indicating that there is no pointer_field in the payload. Refer to 2.4.4.1 and 2.4.4.2. This also applies to private streams of
+    * stream_type 5 (refer to Table 2-29).
+    */
+    if (packet->payload_unit_start_indicator) {
+        if (!stream->require(1)) {
+            ret = ERROR_STREAM_CASTER_TS_AF;
+            srs_error("ts: demux PSI failed. ret=%d", ret);
+            return ret;
+        }
+        pointer_field = stream->read_1bytes();
+    }
+
+    return ret;
+}
+
+SrsTsPayloadPAT::SrsTsPayloadPAT(SrsTsPacket* p) : SrsTsPayloadPSI(p)
+{
+    nb_programs = 0;
+    programs = NULL;
+}
+
+SrsTsPayloadPAT::~SrsTsPayloadPAT()
+{
+    srs_freep(programs);
+}
+
+int SrsTsPayloadPAT::decode(SrsStream* stream)
+{
+    int ret = ERROR_SUCCESS;
+
+    if ((ret = SrsTsPayloadPSI::decode(stream)) != ERROR_SUCCESS) {
+        return ret;
+    }
+
+    // atleast 8B without programs and crc32
+    if (!stream->require(8)) {
+        ret = ERROR_STREAM_CASTER_TS_AF;
+        srs_error("ts: demux PAT failed. ret=%d", ret);
+        return ret;
+    }
+    // 1B
+    table_id = (SrsTsPsiId)stream->read_1bytes();
+    
+    // 2B
+    section_length = stream->read_2bytes();
+    
+    section_syntax_indicator = (section_length >> 15) & 0x01;
+    const0_value = (section_length >> 14) & 0x01;
+    section_length &= 0x0FFF;
+
+    if (!stream->require(section_length)) {
+        ret = ERROR_STREAM_CASTER_TS_AF;
+        srs_error("ts: demux PAT section failed. ret=%d", ret);
+        return ret;
+    }
+    int pos = stream->pos();
+
+    // 2B
+    transport_stream_id = stream->read_2bytes();
+    
+    // 1B
+    current_next_indicator = stream->read_1bytes();
+    
+    version_number = (current_next_indicator >> 1) & 0x1F;
+    current_next_indicator &= 0x01;
+
+    // TODO: FIXME: check the indicator.
+    
+    // 1B
+    section_number = stream->read_1bytes();
+    // 1B
+    last_section_number = stream->read_1bytes();
+
+    // multiple 4B program data.
+    int program_bytes = section_length - 4 - (stream->pos() - pos);
+    nb_programs = program_bytes / 4;
+    if (nb_programs > 0) {
+        srs_freep(programs);
+        programs = new SrsTsPayloadPATProgram[nb_programs];
+
+        for (int i = 0; i < nb_programs; i++) {
+            SrsTsPayloadPATProgram* program = programs + i;
+
+            int tmpv = stream->read_4bytes();
+            program->number = (int16_t)((tmpv >> 16) & 0xFFFF);
+            program->pid = (int16_t)(tmpv & 0x1FFF);
+        }
+    }
+    
+    // 4B
+    if (!stream->require(4)) {
+        ret = ERROR_STREAM_CASTER_TS_AF;
+        srs_error("ts: demux PAT crc32 failed. ret=%d", ret);
+        return ret;
+    }
+    CRC_32 = stream->read_4bytes();
+
+    // TODO: FIXME: verfy crc32.
+
+    return ret;
+}
+
 SrsTSMuxer::SrsTSMuxer(SrsFileWriter* w)
 {
     writer = w;

trunk/src/kernel/srs_kernel_ts.hpp

@@ -42,6 +42,7 @@ class SrsAvcAacCodec;
 class SrsCodecSample;
 class SrsSimpleBuffer;
 class SrsTsAdaptationField;
+class SrsTsPayload;
 
 // Transport Stream packets are 188 bytes in length.
 #define SRS_TS_PACKET_SIZE          188
@@ -108,6 +109,22 @@ enum SrsTsAdaptationFieldType
     SrsTsAdaptationFieldTypeBoth          = 0x03,
 };
 
+/**
+* the context of ts, to decode the ts stream.
+*/
+class SrsTsContext
+{
+public:
+    SrsTsContext();
+    virtual ~SrsTsContext();
+public:
+    /**
+    * the stream contains only one ts packet.
+    * @remark we will consume all bytes in stream.
+    */
+    virtual int decode(SrsStream* stream);
+};
+
 /**
 * the packet in ts stream,
 * 2.4.3.2 Transport Stream packet layer, hls-mpeg-ts-iso13818-1.pdf, page 36
@@ -203,14 +220,11 @@ public:
     u_int8_t continuity_counter; //4bits
 private:
     SrsTsAdaptationField* adaptation_field;
+    SrsTsPayload* payload;
 public:
     SrsTsPacket();
     virtual ~SrsTsPacket();
 public:
-    /**
-    * the stream contains only one ts packet.
-    * @remark we will consume all bytes in stream.
-    */
     virtual int decode(SrsStream* stream);
 };
 
@@ -515,6 +529,189 @@ public:
     virtual int decode(SrsStream* stream);
 };
 
+/**
+* 2.4.4.4 Table_id assignments, hls-mpeg-ts-iso13818-1.pdf, page 62
+* The table_id field identifies the contents of a Transport Stream PSI section as shown in Table 2-26.
+*/
+enum SrsTsPsiId
+{
+    // program_association_section
+    SrsTsPsiIdPas               = 0x00,
+    // conditional_access_section (CA_section)
+    SrsTsPsiIdCas               = 0x01,
+    // TS_program_map_section
+    SrsTsPsiIdPms               = 0x02,
+    // TS_description_section
+    SrsTsPsiIdDs                = 0x03,
+    // ISO_IEC_14496_scene_description_section
+    SrsTsPsiIdSds               = 0x04,
+    // ISO_IEC_14496_object_descriptor_section
+    SrsTsPsiIdOds               = 0x05,
+    // ITU-T Rec. H.222.0 | ISO/IEC 13818-1 reserved
+    SrsTsPsiIdIso138181Start    = 0x06,
+    SrsTsPsiIdIso138181End      = 0x37,
+    // Defined in ISO/IEC 13818-6
+    SrsTsPsiIdIso138186Start    = 0x38,
+    SrsTsPsiIdIso138186End      = 0x3F,
+    // User private
+    SrsTsPsiIdUserStart         = 0x40,
+    SrsTsPsiIdUserEnd           = 0xFE,
+    // forbidden
+    SrsTsPsiIdForbidden         = 0xFF,
+};
+
+/**
+* the program of PAT of PSI ts packet.
+*/
+class SrsTsPayloadPATProgram
+{
+public:
+    // 4B
+    /**
+    * Program_number is a 16-bit field. It specifies the program to which the program_map_PID is
+    * applicable. When set to 0x0000, then the following PID reference shall be the network PID. For all other cases the value
+    * of this field is user defined. This field shall not take any single value more than once within one version of the Program
+    * Association Table.
+    */
+    int16_t number; // 16bits
+    // reserved 3bits
+    /**
+    * program_map_PID/network_PID 13bits
+    * network_PID ¨C The network_PID is a 13-bit field, which is used only in conjunction with the value of the
+    * program_number set to 0x0000, specifies the PID of the Transport Stream packets which shall contain the Network
+    * Information Table. The value of the network_PID field is defined by the user, but shall only take values as specified in
+    * Table 2-3. The presence of the network_PID is optional.
+    */
+    int16_t pid;
+public:
+    SrsTsPayloadPATProgram();
+    virtual ~SrsTsPayloadPATProgram();
+};
+
+/**
+* the payload of ts packet, can be PES or PSI payload.
+*/
+class SrsTsPayload
+{
+protected:
+    SrsTsPacket* packet;
+public:
+    SrsTsPayload(SrsTsPacket* p);
+    virtual ~SrsTsPayload();
+public:
+    virtual int decode(SrsStream* stream) = 0;
+};
+
+/**
+* the PSI payload of ts packet.
+* 2.4.4 Program specific information, hls-mpeg-ts-iso13818-1.pdf, page 59
+*/
+class SrsTsPayloadPSI : public SrsTsPayload
+{
+public:
+    // 1B
+    /**
+    * This is an 8-bit field whose value shall be the number of bytes, immediately following the pointer_field
+    * until the first byte of the first section that is present in the payload of the Transport Stream packet (so a value of 0x00 in
+    * the pointer_field indicates that the section starts immediately after the pointer_field). When at least one section begins in
+    * a given Transport Stream packet, then the payload_unit_start_indicator (refer to 2.4.3.2) shall be set to 1 and the first
+    * byte of the payload of that Transport Stream packet shall contain the pointer. When no section begins in a given
+    * Transport Stream packet, then the payload_unit_start_indicator shall be set to 0 and no pointer shall be sent in the
+    * payload of that packet.
+    */
+    int8_t pointer_field;
+public:
+    SrsTsPayloadPSI(SrsTsPacket* p);
+    virtual ~SrsTsPayloadPSI();
+public:
+    virtual int decode(SrsStream* stream);
+};
+
+/**
+* the PAT payload of PSI ts packet.
+* 2.4.4.3 Program association Table, hls-mpeg-ts-iso13818-1.pdf, page 61
+* The Program Association Table provides the correspondence between a program_number and the PID value of the
+* Transport Stream packets which carry the program definition. The program_number is the numeric label associated with
+* a program.
+*/
+class SrsTsPayloadPAT : public SrsTsPayloadPSI
+{
+public:
+    // 1B
+    /**
+    * This is an 8-bit field, which shall be set to 0x00 as shown in Table 2-26.
+    */
+    SrsTsPsiId table_id; //8bits
+    
+    // 2B
+    /**
+    * The section_syntax_indicator is a 1-bit field which shall be set to '1'.
+    */
+    int8_t section_syntax_indicator; //1bit
+    /**
+    * const value, must be '0'
+    */
+    int8_t const0_value; //1bit
+    // 2bits reserved. 
+    /**
+    * This is a 12-bit field, the first two bits of which shall be '00'. The remaining 10 bits specify the number
+    * of bytes of the section, starting immediately following the section_length field, and including the CRC. The value in this
+    * field shall not exceed 1021 (0x3FD).
+    */
+    u_int16_t section_length; //12bits
+    
+    // 2B
+    /**
+    * This is a 16-bit field which serves as a label to identify this Transport Stream from any other
+    * multiplex within a network. Its value is defined by the user.
+    */
+    u_int16_t transport_stream_id; //16bits
+    
+    // 1B
+    // 2bits reerverd.
+    /**
+    * This 5-bit field is the version number of the whole Program Association Table. The version number
+    * shall be incremented by 1 modulo 32 whenever the definition of the Program Association Table changes. When the
+    * current_next_indicator is set to '1', then the version_number shall be that of the currently applicable Program Association
+    * Table. When the current_next_indicator is set to '0', then the version_number shall be that of the next applicable Program
+    * Association Table.
+    */
+    int8_t version_number; //5bits
+    /**
+    * A 1-bit indicator, which when set to '1' indicates that the Program Association Table sent is
+    * currently applicable. When the bit is set to '0', it indicates that the table sent is not yet applicable and shall be the next
+    * table to become valid.
+    */
+    int8_t current_next_indicator; //1bit
+    
+    // 1B
+    /**
+    * This 8-bit field gives the number of this section. The section_number of the first section in the
+    * Program Association Table shall be 0x00. It shall be incremented by 1 with each additional section in the Program
+    * Association Table.
+    */
+    u_int8_t section_number; //8bits
+    
+    // 1B
+    /**
+    * This 8-bit field specifies the number of the last section (that is, the section with the highest
+    * section_number) of the complete Program Association Table.
+    */
+    u_int8_t last_section_number; //8bits
+    
+    // multiple 4B program data.
+    int nb_programs;
+    SrsTsPayloadPATProgram* programs;
+    
+    // 4B
+    int32_t CRC_32; //32bits
+public:
+    SrsTsPayloadPAT(SrsTsPacket* p);
+    virtual ~SrsTsPayloadPAT();
+public:
+    virtual int decode(SrsStream* stream);
+};
+
 /**
 * write data from frame(header info) and buffer(data) to ts file.
 * it's a simple object wrapper for utility from nginx-rtmp: SrsMpegtsWriter