Generate NACK (#64)

* C++: Provide the RtpStream with the stream SSRC.
* Assume NACK just for video and depth kinds.
* ...and send NACK to senders.

gulpfile.js

@@ -62,7 +62,8 @@ gulp.task('test:worker', shell.task(
 		`cd worker && ./out/${process.env.MEDIASOUP_BUILDTYPE === 'Debug' ? 'Debug' : 'Release'}/mediasoup-worker-test --invisibles --use-colour=yes`
 	],
 	{
-		verbose : true
+		verbose : true,
+		env     : { DEBUG: '*ABORT* *WARN*' }
 	}
 ));
 

worker/include/Logger.hpp

@@ -77,6 +77,27 @@
 	#define _MS_LOG_DEV_ENABLED false
 #endif
 
+// Usage:
+//   MS_DEBUG_DEV("Leading text "MS_UINT16_TO_BINARY_PATTERN, MS_UINT16_TO_BINARY(value));
+#define MS_UINT16_TO_BINARY_PATTERN "%c%c%c%c%c%c%c%c%c%c%c%c%c%c%c%c"
+#define MS_UINT16_TO_BINARY(value) \
+	((value & 0x8000) ? '1' : '0'), \
+	((value & 0x4000) ? '1' : '0'), \
+	((value & 0x2000) ? '1' : '0'), \
+	((value & 0x1000) ? '1' : '0'), \
+	((value & 0x800) ? '1' : '0'), \
+	((value & 0x400) ? '1' : '0'), \
+	((value & 0x200) ? '1' : '0'), \
+	((value & 0x100) ? '1' : '0'), \
+	((value & 0x80) ? '1' : '0'), \
+	((value & 0x40) ? '1' : '0'), \
+	((value & 0x20) ? '1' : '0'), \
+	((value & 0x10) ? '1' : '0'), \
+	((value & 0x08) ? '1' : '0'), \
+	((value & 0x04) ? '1' : '0'), \
+	((value & 0x02) ? '1' : '0'), \
+	((value & 0x01) ? '1' : '0')
+
 class Logger
 {
 public:

worker/include/RTC/RTCP/FeedbackRtpNack.hpp

@@ -9,7 +9,7 @@
  *
      0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
-0   |              PID              |             BPL               |
+    |              PID              |             BPL               |
     +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
  */
 

worker/include/RTC/RtpReceiver.hpp

@@ -20,7 +20,8 @@ namespace RTC
 	// the corresponding header files.
 	class Transport;
 
-	class RtpReceiver
+	class RtpReceiver :
+		public RtpStreamRecv::Listener
 	{
 	public:
 		/**
@@ -58,6 +59,10 @@ namespace RTC
 	private:
 		void ClearRtpStreams();
 
+	/* Pure virtual methods inherited from RTC::RtpStreamRecv::Listener. */
+	public:
+		virtual void onNackRequired(RTC::RtpStreamRecv* rtpStream, uint16_t seq, uint16_t bitmask) override;
+
 	public:
 		// Passed by argument.
 		uint32_t rtpReceiverId;

worker/include/RTC/RtpStream.hpp

@@ -9,17 +9,23 @@ namespace RTC
 	class RtpStream
 	{
 	public:
-		explicit RtpStream(uint32_t clockRate);
+		explicit RtpStream(uint32_t ssrc, uint32_t clockRate);
 		virtual ~RtpStream();
 
+		uint32_t GetSsrc();
 		virtual bool ReceivePacket(RTC::RtpPacket* packet);
 
 	private:
 		void InitSeq(uint16_t seq);
 		bool UpdateSeq(uint16_t seq);
 
+	/* Pure virtual methods that must be implemented by the subclass. */
+	protected:
+		virtual void onInitSeq() = 0;
+
 	protected:
 		// Given as argument.
+		uint32_t ssrc = 0;
 		uint32_t clockRate = 0;
 		bool started = false; // Whether at least a RTP packet has been received.
 		// https://tools.ietf.org/html/rfc3550#appendix-A.1 stuff.
@@ -34,6 +40,14 @@ namespace RTC
 		// Others.
 		uint32_t max_timestamp = 0; // Highest timestamp seen.
 	};
+
+	/* Inline instance methods. */
+
+	inline
+	uint32_t RtpStream::GetSsrc()
+	{
+		return this->ssrc;
+	}
 }
 
 #endif

worker/include/RTC/RtpStreamRecv.hpp

@@ -12,22 +12,39 @@ namespace RTC
 		public RtpStream
 	{
 	public:
-		explicit RtpStreamRecv(uint32_t clockRate);
+		class Listener
+		{
+		public:
+			virtual void onNackRequired(RTC::RtpStreamRecv* rtpStream, uint16_t seq, uint16_t bitmask) = 0;
+		};
+
+	public:
+		explicit RtpStreamRecv(Listener* listener, uint32_t ssrc, uint32_t clockRate, bool useNack);
 		virtual ~RtpStreamRecv();
 
 		Json::Value toJson();
-		virtual bool ReceivePacket(RTC::RtpPacket* packet);
+		virtual bool ReceivePacket(RTC::RtpPacket* packet) override;
 		RTC::RTCP::ReceiverReport* GetRtcpReceiverReport();
 		void ReceiveRtcpSenderReport(RTC::RTCP::SenderReport* report);
 
 	private:
 		void CalculateJitter(uint32_t rtpTimestamp);
+		void MayTriggerNack(RTC::RtpPacket* packet);
+		void ResetNack();
+
+	/* Pure virtual methods inherited from RtpStream. */
+	protected:
+		virtual void onInitSeq() override;
 
 	private:
+		// Passed by argument.
+		Listener* listener = nullptr;
+		bool useNack = false;
 		uint32_t last_sr_timestamp = 0; // The middle 32 bits out of 64 in the NTP timestamp received in the most recent sender report.
 		uint64_t last_sr_received = 0; // Wallclock time representing the most recent sender report arrival.
 		uint32_t transit = 0; // Relative trans time for prev pkt.
 		uint32_t jitter = 0; // Estimated jitter.
+		uint32_t last_seq32 = 0; // Extended seq number of last valid packet.
 	};
 }
 

worker/include/RTC/RtpStreamSend.hpp

@@ -26,18 +26,22 @@ namespace RTC
 		};
 
 	public:
-		RtpStreamSend(uint32_t clockRate, size_t bufferSize);
+		RtpStreamSend(uint32_t ssrc, uint32_t clockRate, size_t bufferSize);
 		virtual ~RtpStreamSend();
 
 		Json::Value toJson();
-		bool ReceivePacket(RTC::RtpPacket* packet);
+		bool ReceivePacket(RTC::RtpPacket* packet) override;
 		void RequestRtpRetransmission(uint16_t seq, uint16_t bitmask, std::vector<RTC::RtpPacket*>& container);
 		RTC::RTCP::SenderReport* GetRtcpSenderReport(uint64_t now);
 
 	private:
 		void ClearBuffer();
 		void StorePacket(RTC::RtpPacket* packet);
 
+	/* Pure virtual methods inherited from RtpStream. */
+	protected:
+		virtual void onInitSeq() override;
+
 	private:
 		std::vector<StorageItem> storage;
 		typedef std::list<BufferItem> Buffer;

worker/include/common.hpp

@@ -2,7 +2,7 @@
 #define MS_COMMON_HPP
 
 #include <memory> // std::addressof()
-#include <algorithm> // std::transform(), std::find()
+#include <algorithm> // std::transform(), std::find(), std::min(), std::max()
 #include <cstddef> // size_t
 #include <cstdint> // uint8_t, etc
 #include <cinttypes> // PRIu64, etc

worker/mediasoup-worker.gyp

@@ -230,7 +230,7 @@
     },
     {
       'target_name': 'mediasoup-worker-test',
-      'defines': [ 'MS_LOG_STD' ],
+      'defines': [ 'MS_TEST', 'MS_LOG_STD' ],
       'sources':
       [
         # C++ source files
@@ -239,6 +239,7 @@
         'test/test-rtp.cpp',
         'test/test-rtcp.cpp',
         'test/test-bitrate.cpp',
+        'test/test-rtpstreamrecv.cpp',
         # C++ include files
         'test/catch.hpp',
         'test/helpers.hpp'

worker/src/RTC/RTCP/FeedbackRtpNack.cpp

@@ -4,6 +4,7 @@
 #include "RTC/RTCP/FeedbackRtpNack.hpp"
 #include "Logger.hpp"
 #include <cstring>
+#include <bitset> // std::bitset()
 
 namespace RTC { namespace RTCP
 {
@@ -50,9 +51,11 @@ namespace RTC { namespace RTCP
 	{
 		MS_TRACE();
 
+		std::bitset<16> nack_bitset(this->GetLostPacketBitmask());
+
 		MS_DUMP("<NackItem>");
 		MS_DUMP("  pid : %" PRIu16, this->GetPacketId());
-		MS_DUMP("  bpl : %" PRIu16, this->GetLostPacketBitmask());
+		MS_DUMP("  bpl : %s", nack_bitset.to_string().c_str());
 		MS_DUMP("</NackItem>");
 	}
 }}

worker/src/RTC/RtpReceiver.cpp

@@ -3,7 +3,8 @@
 
 #include "RTC/RtpReceiver.hpp"
 #include "RTC/Transport.hpp"
-#include "Utils.hpp"
+#include "RTC/RTCP/FeedbackRtp.hpp"
+#include "RTC/RTCP/FeedbackRtpNack.hpp"
 #include "MediaSoupError.hpp"
 #include "Logger.hpp"
 
@@ -69,10 +70,10 @@ namespace RTC
 		static const Json::StaticString k_rtpRawEventEnabled("rtpRawEventEnabled");
 		static const Json::StaticString k_rtpObjectEventEnabled("rtpObjectEventEnabled");
 		static const Json::StaticString k_rtpStreams("rtpStreams");
-		static const Json::StaticString k_ssrc("ssrc");
 		static const Json::StaticString k_rtpStream("rtpStream");
 
 		Json::Value json(Json::objectValue);
+		Json::Value json_rtpStreams(Json::arrayValue);
 
 		json[k_rtpReceiverId] = (Json::UInt)this->rtpReceiverId;
 
@@ -89,19 +90,13 @@ namespace RTC
 
 		json[k_rtpObjectEventEnabled] = this->rtpObjectEventEnabled;
 
-		json[k_rtpStreams] = Json::arrayValue;
-
 		for (auto& kv : this->rtpStreams)
 		{
-			auto ssrc = kv.first;
 			auto rtpStream = kv.second;
-			Json::Value json_rtpStream(Json::objectValue);
-
-			json_rtpStream[k_ssrc] = (Json::UInt)ssrc;
-			json_rtpStream[k_rtpStream] = rtpStream->toJson();
 
-			json[k_rtpStreams].append(json_rtpStream);
+			json_rtpStreams.append(rtpStream->toJson());
 		}
+		json[k_rtpStreams] = json_rtpStreams;
 
 		return json;
 	}
@@ -199,9 +194,11 @@ namespace RTC
 
 					// Get the clock rate of the stream/encoding.
 					uint32_t streamClockRate = this->rtpParameters->GetClockRateForEncoding(encoding);
+					// TODO: Let's assume that, if video, NACK is negotiated. Must do this better.
+					bool useNack = (this->kind != RTC::Media::Kind::AUDIO);
 
 					// Create a RtpStreamRecv for receiving a media stream.
-					this->rtpStreams[ssrc] = new RTC::RtpStreamRecv(streamClockRate);
+					this->rtpStreams[ssrc] = new RTC::RtpStreamRecv(this, ssrc, streamClockRate, useNack);
 				}
 
 				break;
@@ -324,11 +321,10 @@ namespace RTC
 
 		for (auto& kv : this->rtpStreams)
 		{
-			auto ssrc = kv.first;
 			auto rtpStream = kv.second;
 			RTC::RTCP::ReceiverReport* report = rtpStream->GetRtcpReceiverReport();
 
-			report->SetSsrc(ssrc);
+			report->SetSsrc(rtpStream->GetSsrc());
 			packet->AddReceiverReport(report);
 		}
 
@@ -388,4 +384,17 @@ namespace RTC
 
 		this->rtpStreams.clear();
 	}
+
+	void RtpReceiver::onNackRequired(RTC::RtpStreamRecv* rtpStream, uint16_t seq, uint16_t bitmask)
+	{
+		if (!this->transport)
+			return;
+
+		RTC::RTCP::FeedbackRtpNackPacket packet(0, rtpStream->GetSsrc());
+		RTC::RTCP::NackItem* nackItem = new RTC::RTCP::NackItem(seq, bitmask);
+
+		packet.AddItem(nackItem);
+		packet.Serialize(RtpReceiver::rtcpBuffer);
+		this->transport->SendRtcpPacket(&packet);
+	}
 }

worker/src/RTC/RtpSender.cpp

@@ -258,8 +258,7 @@ namespace RTC
 
 			// NOTE: We assume a single stream/encoding when sending to remote peers.
 			auto encoding = this->rtpParameters->encodings[0];
-
-			// Set the RtpStreamSend.
+			uint32_t ssrc = encoding.ssrc;
 			uint32_t streamClockRate = this->rtpParameters->GetClockRateForEncoding(encoding);
 
 			// Create a RtpStreamSend for sending a single media stream.
@@ -268,15 +267,15 @@ namespace RTC
 				case RTC::Media::Kind::VIDEO:
 				case RTC::Media::Kind::DEPTH:
 				{
-					// Buffer up to 100 packets.
-					this->rtpStream = new RTC::RtpStreamSend(streamClockRate, 100);
+					// Buffer up to N packets.
+					this->rtpStream = new RTC::RtpStreamSend(ssrc, streamClockRate, 200);
 					break;
 				}
 
 				case RTC::Media::Kind::AUDIO:
 				{
 					// No buffer for audio streams.
-					this->rtpStream = new RTC::RtpStreamSend(streamClockRate, 0);
+					this->rtpStream = new RTC::RtpStreamSend(ssrc, streamClockRate, 0);
 					break;
 				}
 

worker/src/RTC/RtpStream.cpp

@@ -15,7 +15,8 @@ namespace RTC
 {
 	/* Instance methods. */
 
-	RtpStream::RtpStream(uint32_t clockRate) :
+	RtpStream::RtpStream(uint32_t ssrc, uint32_t clockRate) :
+		ssrc(ssrc),
 		clockRate(clockRate)
 	{
 		MS_TRACE();
@@ -45,7 +46,10 @@ namespace RTC
 		// If not a valid packet ignore it.
 		if (!UpdateSeq(seq))
 		{
-			MS_WARN_TAG(rtp, "invalid packet [seq:%" PRIu16 "]", packet->GetSequenceNumber());
+			if (!this->probation)
+			{
+				MS_WARN_TAG(rtp, "invalid packet [seq:%" PRIu16 "]", packet->GetSequenceNumber());
+			}
 
 			return false;
 		}
@@ -71,6 +75,9 @@ namespace RTC
 		this->expected_prior = 0;
 		// Also reset the highest seen RTP timestamp.
 		this->max_timestamp = 0;
+
+		// Call the onInitSeq method of the child.
+		onInitSeq();
 	}
 
 	bool RtpStream::UpdateSeq(uint16_t seq)