Add support for RTP input in srt-live-transmit

apps/srt-live-transmit.cpp

@@ -526,11 +526,13 @@ int main(int argc, char** argv)
                     }
                     break;
                 case UriParser::UDP:
+                case UriParser::RTP:
                     if (srt_epoll_add_ssock(pollid,
                         src->GetSysSocket(), &events))
                     {
-                        cerr << "Failed to add UDP source to poll, "
-                            << src->GetSysSocket() << endl;
+                        cerr << "Failed to add " << src->uri.proto()
+                            << " source to poll, " << src->GetSysSocket()
+                            << endl;
                         return 1;
                     }
                     break;

apps/transmitmedia.cpp

@@ -985,6 +985,7 @@ class UdpCommon
 
 class UdpSource: public Source, public UdpCommon
 {
+protected:
     bool eof = true;
 public:
 
@@ -1082,6 +1083,74 @@ template <> struct Udp<Target> { typedef UdpTarget type; };
 template <class Iface>
 Iface* CreateUdp(const string& host, int port, const map<string,string>& par) { return new typename Udp<Iface>::type (host, port, par); }
 
+class RtpSource: public UdpSource
+{
+    // for now, make no effort to parse the header, just assume it is always
+    // fixed length and either a user-configurable value, or twelve bytes.
+    const int MINIMUM_RTP_HEADER_SIZE = 12;
+    int bytes_to_skip = MINIMUM_RTP_HEADER_SIZE;
+public:
+    RtpSource(string host, int port, const map<string,string>& attr) :
+        UdpSource { host, port, attr }
+        {
+            if (attr.count("rtpheadersize"))
+            {
+                const int header_size = stoi(attr.at("rtpheadersize"), 0, 0);
+                if (header_size < MINIMUM_RTP_HEADER_SIZE)
+                {
+                    cerr << "Invalid RTP header size provided: " << header_size
+                        << ", minimum allowed is " << MINIMUM_RTP_HEADER_SIZE
+                        << endl;
+                    throw invalid_argument("Invalid RTP header size");
+                }
+                bytes_to_skip = header_size;
+            }
+        }
+
+    int Read(size_t chunk, MediaPacket& pkt, ostream & ignored SRT_ATR_UNUSED = cout) override
+    {
+        const int length = UdpSource::Read(chunk, pkt);
+
+        if (length < 1 || !bytes_to_skip)
+        {
+            // something went wrong, or we're not skipping bytes for some
+            // reason, just return the length read via the base method
+            return length;
+        }
+
+        // we got some data and we're supposed to skip some of it
+        // check there's enough bytes for our intended skip
+        if (length < bytes_to_skip)
+        {
+            // something went wrong here
+            cerr << "RTP packet too short (" << length
+                << " bytes) to remove headers (needed "
+                << bytes_to_skip << ")" << endl;
+            throw std::runtime_error("Unexpected RTP packet length");
+        }
+
+        pkt.payload.erase(
+            pkt.payload.begin(),
+            pkt.payload.begin() + bytes_to_skip
+        );
+
+        return length - bytes_to_skip;
+    }
+};
+
+class RtpTarget : public UdpTarget {
+public:
+    RtpTarget(string host, int port, const map<string,string>& attr ) :
+        UdpTarget { host, port, attr } {}
+};
+
+template <class Iface> struct Rtp;
+template <> struct Rtp<Source> { typedef RtpSource type; };
+template <> struct Rtp<Target> { typedef RtpTarget type; };
+
+template <class Iface>
+Iface* CreateRtp(const string& host, int port, const map<string,string>& par) { return new typename Rtp<Iface>::type (host, port, par); }
+
 template<class Base>
 inline bool IsOutput() { return false; }
 
@@ -1141,6 +1210,20 @@ extern unique_ptr<Base> CreateMedium(const string& uri)
         ptr.reset( CreateUdp<Base>(u.host(), iport, u.parameters()) );
         break;
 
+    case UriParser::RTP:
+        if (IsOutput<Base>())
+        {
+            cerr << "RTP not supported as an output\n";
+            throw invalid_argument("Invalid output protocol: RTP");
+        }
+        iport = atoi(u.port().c_str());
+        if ( iport < 1024 )
+        {
+            cerr << "Port value invalid: " << iport << " - must be >=1024\n";
+            throw invalid_argument("Invalid port number");
+        }
+        ptr.reset( CreateRtp<Base>(u.host(), iport, u.parameters()) );
+        break;
     }
 
     if (ptr.get())

docs/apps/srt-live-transmit.md

@@ -17,6 +17,7 @@ The following medium types are handled by `srt-live-transmit`:
 
 - SRT - use SRT for reading or writing, in listener, caller or rendezvous mode, with possibly additional parameters
 - UDP - read or write the given UDP address (also multicast)
+- RTP - read RTP from the given address (also multicast)
 - Local file - read or store the stream into the file
 - Process's pipeline - use the process's `stdin` and `stdout` standard streams
 
@@ -86,6 +87,7 @@ The applications supports the following schemes:
 
 - `file` - for file or standard input and output
 - `udp` - UDP output (unicast and multicast)
+- `rtp` - RTP input (unicast and multicast)
 - `srt` - SRT connection
 
 Note that this application doesn't support file as a medium, but this
@@ -183,6 +185,25 @@ instead of `IP_ADD_MEMBERSHIP` and the value is set to `imr_sourceaddr` field.
 Explanations for the symbols and terms used above can be found in POSIX
 manual pages, like `ip(7)` and on Microsoft docs pages under `IPPROTO_IP`.
 
+### Medium: RTP
+
+RTP is supported for input only.
+
+All URI parameters described in the [Medium: UDP](#medium-udp) section above
+also apply to RTP. A further RTP-specific option is available as an URI
+parameter:
+
+- **rtpheadersize**: sets the number of bytes to drop from the beginning of
+each received packet. Defaults to 12 if not provided. Minimum value is 12.
+
+A length of **rtpheadersize** bytes will always be dropped. If you wish to pass
+the entire packet, including RTP header, to the output medium, you should
+instead specify UDP as the input medium.
+
+> NOTE: No effort is made in the initial implementation to attempt to parse
+the RTP headers in any way eg for validation, reordering, extracting timing,
+length detection of checking.
+
 ### Medium: SRT
 
 Most important about SRT is that it can be either input or output and in