Add periodic Pli handler

erizo/src/erizo/MediaStream.cpp

@@ -34,6 +34,7 @@
 #include "rtp/LayerBitrateCalculationHandler.h"
 #include "rtp/QualityFilterHandler.h"
 #include "rtp/QualityManager.h"
+#include "rtp/PeriodicPliHandler.h"
 #include "rtp/PliPriorityHandler.h"
 #include "rtp/PliPacerHandler.h"
 #include "rtp/RtpPaddingGeneratorHandler.h"
@@ -67,7 +68,9 @@ MediaStream::MediaStream(std::shared_ptr<Worker> worker,
     bitrate_from_max_quality_layer_{0},
     video_bitrate_{0},
     random_generator_{random_device_()},
-    target_padding_bitrate_{0} {
+    target_padding_bitrate_{0},
+    periodic_keyframes_requested_{false},
+    periodic_keyframe_interval_{0} {
   if (is_publisher) {
     setVideoSinkSSRC(kDefaultVideoSinkSSRC);
     setAudioSinkSSRC(kDefaultAudioSinkSSRC);
@@ -399,6 +402,7 @@ void MediaStream::initializePipeline() {
   pipeline_->addFront(std::make_shared<RtpTrackMuteHandler>());
   pipeline_->addFront(std::make_shared<RtpSlideShowHandler>());
   pipeline_->addFront(std::make_shared<RtpPaddingGeneratorHandler>());
+  pipeline_->addFront(std::make_shared<PeriodicPliHandler>());
   pipeline_->addFront(std::make_shared<PliPriorityHandler>());
   pipeline_->addFront(std::make_shared<PliPacerHandler>());
   pipeline_->addFront(std::make_shared<RtpPaddingRemovalHandler>());
@@ -604,6 +608,13 @@ void MediaStream::sendPLIToFeedback() {
       this->getVideoSourceSSRC()));
   }
 }
+
+void MediaStream::setPeriodicKeyframeRequests(bool activate, uint32_t interval) {
+  ELOG_DEBUG("%s message: settingPeriodicKeyframes, activate: %u, interval, %u", activate, interval);
+  periodic_keyframes_requested_ = activate;
+  periodic_keyframe_interval_ = interval;
+  notifyUpdateToHandlers();
+}
 // changes the outgoing payload type for in the given data packet
 void MediaStream::sendPacketAsync(std::shared_ptr<DataPacket> packet) {
   if (!sending_) {

erizo/src/erizo/MediaStream.h

@@ -87,6 +87,7 @@ class MediaStream: public MediaSink, public MediaSource, public FeedbackSink,
   void sendPLIToFeedback();
   void setQualityLayer(int spatial_layer, int temporal_layer);
   void enableSlideShowBelowSpatialLayer(bool enabled, int spatial_layer);
+  void setPeriodicKeyframeRequests(bool activate, uint32_t interval_in_ms = 0);
 
   WebRTCEvent getCurrentState();
 
@@ -142,6 +143,9 @@ class MediaStream: public MediaSink, public MediaSource, public FeedbackSink,
 
   virtual bool isSlideShowModeEnabled() { return slide_show_mode_; }
 
+  virtual bool isRequestingPeriodicKeyframes() { return periodic_keyframes_requested_; }
+  virtual uint32_t getPeriodicKeyframesRequesInterval() { return periodic_keyframe_interval_; }
+
   virtual bool isSimulcast() { return simulcast_; }
   void setSimulcast(bool simulcast) { simulcast_ = simulcast; }
 
@@ -230,6 +234,9 @@ class MediaStream: public MediaSink, public MediaSource, public FeedbackSink,
   std::random_device random_device_;
   std::mt19937 random_generator_;
   uint64_t target_padding_bitrate_;
+  bool periodic_keyframes_requested_;
+  uint32_t periodic_keyframe_interval_;
+
  protected:
   std::shared_ptr<SdpInfo> remote_sdp_;
 };

erizo/src/erizo/rtp/PeriodicPliHandler.cpp

@@ -0,0 +1,96 @@
+#include "rtp/PeriodicPliHandler.h"
+
+#include "rtp/RtpUtils.h"
+#include "./MediaDefinitions.h"
+#include "./MediaStream.h"
+
+namespace erizo {
+
+DEFINE_LOGGER(PeriodicPliHandler, "rtp.PeriodicPliHandler");
+
+PeriodicPliHandler::PeriodicPliHandler(std::shared_ptr<erizo::Clock> the_clock)
+    : enabled_{true}, stream_{nullptr}, clock_{the_clock},
+      video_sink_ssrc_{0}, video_source_ssrc_{0}, keyframes_received_in_interval_{0},
+      requested_periodic_plis_{false}, has_scheduled_pli_{false} {}
+
+void PeriodicPliHandler::enable() {
+  enabled_ = true;
+}
+
+void PeriodicPliHandler::disable() {
+  enabled_ = false;
+}
+
+void PeriodicPliHandler::notifyUpdate() {
+  auto pipeline = getContext()->getPipelineShared();
+  if (pipeline && !stream_) {
+    stream_ = pipeline->getService<MediaStream>().get();
+    video_sink_ssrc_ = stream_->getVideoSinkSSRC();
+    video_source_ssrc_ = stream_->getVideoSourceSSRC();
+  }
+  // check if is actuve and start pliReuqests or update the variable so it stops on its own
+
+  updateInterval(stream_->isRequestingPeriodicKeyframes(), stream_->getPeriodicKeyframesRequesInterval());
+}
+
+void PeriodicPliHandler::updateInterval(bool active, uint32_t interval_ms) {
+  requested_periodic_plis_ = active;
+  requested_interval_ = std::chrono::milliseconds(interval_ms);
+  if (enabled_ && requested_periodic_plis_ && !has_scheduled_pli_) {
+    ELOG_DEBUG("%s, message: Updating interval, requested_periodic_plis_: %u, interval: %u", stream_->toLog(),
+        requested_periodic_plis_, requested_interval_);
+    scheduleNextPli(requested_interval_);
+    has_scheduled_pli_ = true;
+  }
+}
+
+void PeriodicPliHandler::read(Context *ctx, std::shared_ptr<DataPacket> packet) {
+  if (enabled_ && packet->is_keyframe) {
+    ELOG_DEBUG("%s, message: Received Keyframe, total in interval %u", stream_->toLog(),
+        keyframes_received_in_interval_);
+    keyframes_received_in_interval_++;  // the pli counter because we have keyframe
+  }
+  ctx->fireRead(std::move(packet));
+}
+
+void PeriodicPliHandler::write(Context *ctx, std::shared_ptr<DataPacket> packet) {
+  ctx->fireWrite(std::move(packet));
+}
+
+void PeriodicPliHandler::sendPLI() {
+  getContext()->fireWrite(RtpUtils::createPLI(video_source_ssrc_, video_sink_ssrc_, HIGH_PRIORITY));
+}
+
+
+void PeriodicPliHandler::scheduleNextPli(duration next_pli_time) {
+  if (!enabled_) {
+    return;
+  }
+  ELOG_INFO("%s, message: scheduleNextKeyframe, keyframes_received_in_interval_: %u, next_keyframe %u",
+      stream_->toLog(), keyframes_received_in_interval_, ClockUtils::durationToMs(next_pli_time));;
+  std::weak_ptr<PeriodicPliHandler> weak_this = shared_from_this();
+  stream_->getWorker()->scheduleFromNow([weak_this] {
+    if (auto this_ptr = weak_this.lock()) {
+      if (this_ptr->requested_periodic_plis_) {
+        ELOG_DEBUG("%s, message: Maybe Sending PLI, keyframes_received_in_interval_: %u",
+            this_ptr->stream_->toLog(), this_ptr->keyframes_received_in_interval_);
+        if (this_ptr->keyframes_received_in_interval_ == 0) {
+          ELOG_DEBUG("%s, message: Will send PLI, keyframes_received_in_interval_: %u",
+              this_ptr->stream_->toLog(), this_ptr->keyframes_received_in_interval_);
+          this_ptr->sendPLI();
+        }
+        this_ptr->keyframes_received_in_interval_ = 0;
+        ELOG_DEBUG("%s, message: Scheduling next pli in : %u",
+            this_ptr->stream_->toLog(), ClockUtils::durationToMs(this_ptr->requested_interval_));
+        this_ptr->scheduleNextPli(this_ptr->requested_interval_);
+      } else {
+        ELOG_DEBUG("%s, message: Not scheduling more PLIs: %u",
+            this_ptr->stream_->toLog(), this_ptr->keyframes_received_in_interval_);
+        this_ptr->has_scheduled_pli_ = false;
+        this_ptr->sendPLI();
+      }
+    }
+  }, next_pli_time);
+}
+}  // namespace erizo
+

erizo/src/erizo/rtp/PeriodicPliHandler.h

@@ -0,0 +1,53 @@
+#ifndef ERIZO_SRC_ERIZO_RTP_PERIODICPLIHANDLER_H_
+#define ERIZO_SRC_ERIZO_RTP_PERIODICPLIHANDLER_H_
+
+#include <string>
+
+#include "./logger.h"
+#include "pipeline/Handler.h"
+#include "thread/Worker.h"
+#include "lib/Clock.h"
+
+namespace erizo {
+
+class MediaStream;
+
+class PeriodicPliHandler: public Handler, public std::enable_shared_from_this<PeriodicPliHandler> {
+  DECLARE_LOGGER();
+
+ public:
+  explicit PeriodicPliHandler(std::shared_ptr<erizo::Clock> the_clock = std::make_shared<SteadyClock>());
+
+  void enable() override;
+  void disable() override;
+
+  std::string getName() override {
+    return "periodic-pli";
+  }
+
+  void read(Context *ctx, std::shared_ptr<DataPacket> packet) override;
+  void write(Context *ctx, std::shared_ptr<DataPacket> packet) override;
+  void notifyUpdate() override;
+  void updateInterval(bool active, uint32_t interval_ms);
+
+ private:
+  void scheduleNextPli(duration next_pli_time);
+  void sendPLI();
+
+ private:
+  bool enabled_;
+  MediaStream* stream_;
+  std::shared_ptr<erizo::Clock> clock_;
+  uint32_t video_sink_ssrc_;
+  uint32_t video_source_ssrc_;
+  uint32_t keyframes_received_in_interval_;
+  duration requested_interval_;
+  bool requested_periodic_plis_;
+
+  bool has_scheduled_pli_;
+};
+
+}  // namespace erizo
+
+#endif  // ERIZO_SRC_ERIZO_RTP_PERIODICPLIHANDLER_H_
+

erizo/src/test/rtp/PeriodicPliHandlerTest.cpp

@@ -0,0 +1,124 @@
+
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+#include <rtp/PeriodicPliHandler.h>
+#include <rtp/RtpHeaders.h>
+#include <MediaDefinitions.h>
+#include <WebRtcConnection.h>
+
+#include <queue>
+#include <string>
+#include <vector>
+
+#include "../utils/Mocks.h"
+#include "../utils/Tools.h"
+#include "../utils/Matchers.h"
+
+using ::testing::_;
+using ::testing::IsNull;
+using ::testing::Args;
+using ::testing::Return;
+using erizo::DataPacket;
+using erizo::packetType;
+using erizo::AUDIO_PACKET;
+using erizo::VIDEO_PACKET;
+using erizo::IceConfig;
+using erizo::RtpMap;
+using erizo::PeriodicPliHandler;
+using erizo::SimulatedClock;
+using erizo::WebRtcConnection;
+using erizo::Pipeline;
+using erizo::InboundHandler;
+using erizo::OutboundHandler;
+using erizo::Worker;
+using std::queue;
+
+constexpr int kArbitraryKeyframePeriodMs = 1000;
+
+
+class PeriodicPliHandlerTest : public erizo::HandlerTest {
+ public:
+  PeriodicPliHandlerTest() {}
+
+ protected:
+  void setHandler() {
+    periodic_pli_handler = std::make_shared<PeriodicPliHandler>(simulated_clock);
+    pipeline->addBack(periodic_pli_handler);
+  }
+
+  std::shared_ptr<PeriodicPliHandler> periodic_pli_handler;
+};
+
+
+TEST_F(PeriodicPliHandlerTest, basicBehaviourShouldReadPackets) {
+    auto packet = erizo::PacketTools::createDataPacket(erizo::kArbitrarySeqNumber, AUDIO_PACKET);
+
+    EXPECT_CALL(*reader.get(), read(_, _)).
+      With(Args<1>(erizo::RtpHasSequenceNumber(erizo::kArbitrarySeqNumber))).Times(1);
+    pipeline->read(packet);
+}
+
+TEST_F(PeriodicPliHandlerTest, basicBehaviourShouldWritePackets) {
+    auto packet = erizo::PacketTools::createDataPacket(erizo::kArbitrarySeqNumber, AUDIO_PACKET);
+
+    EXPECT_CALL(*writer.get(), write(_, _)).
+      With(Args<1>(erizo::RtpHasSequenceNumber(erizo::kArbitrarySeqNumber))).Times(1);
+    pipeline->write(packet);
+}
+
+TEST_F(PeriodicPliHandlerTest, shouldSendPliEveryInterval) {
+    periodic_pli_handler->updateInterval(true, kArbitraryKeyframePeriodMs);
+    EXPECT_CALL(*writer.get(), write(_, _)).With(Args<1>(erizo::IsPLI())).Times(2);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+}
+
+TEST_F(PeriodicPliHandlerTest, shouldNotSendPliIfDeactivated) {
+    periodic_pli_handler->updateInterval(false, 0);
+    EXPECT_CALL(*writer.get(), write(_, _)).With(Args<1>(erizo::IsPLI())).Times(0);
+
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+}
+
+TEST_F(PeriodicPliHandlerTest, shouldUpdateIntervalIfRequested) {
+    periodic_pli_handler->updateInterval(true, kArbitraryKeyframePeriodMs);
+    EXPECT_CALL(*writer.get(), write(_, _)).With(Args<1>(erizo::IsPLI())).Times(2);
+
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+    periodic_pli_handler->updateInterval(true, 2*kArbitraryKeyframePeriodMs);
+    executeTasksInNextMs(2*kArbitraryKeyframePeriodMs+1);
+}
+
+TEST_F(PeriodicPliHandlerTest, shouldNotSendPliIfKeyframeIsReceivedInPeriod) {
+    auto keyframe = erizo::PacketTools::createVP8Packet(erizo::kArbitrarySeqNumber, true, true);
+
+    EXPECT_CALL(*writer.get(), write(_, _)).With(Args<1>(erizo::IsPLI())).Times(0);
+    EXPECT_CALL(*reader.get(), read(_, _)).
+      With(Args<1>(erizo::RtpHasSequenceNumber(erizo::kArbitrarySeqNumber))).Times(1);
+
+    periodic_pli_handler->updateInterval(true, kArbitraryKeyframePeriodMs);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs/2);
+    pipeline->read(keyframe);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs/2 + 1);
+}
+
+TEST_F(PeriodicPliHandlerTest, shouldSendPliWhenRequestedToStop) {
+    auto keyframe = erizo::PacketTools::createVP8Packet(erizo::kArbitrarySeqNumber, true, true);
+
+    EXPECT_CALL(*writer.get(), write(_, _)).With(Args<1>(erizo::IsPLI())).Times(1);
+
+    periodic_pli_handler->updateInterval(true, kArbitraryKeyframePeriodMs);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs/3);
+    periodic_pli_handler->updateInterval(false, 0);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs*2 + 1);
+}
+
+TEST_F(PeriodicPliHandlerTest, shouldNotSchedulePlisWhenDisabled) {
+    periodic_pli_handler->disable();
+
+    periodic_pli_handler->updateInterval(true, kArbitraryKeyframePeriodMs);
+    EXPECT_CALL(*writer.get(), write(_, _)).With(Args<1>(erizo::IsPLI())).Times(0);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+    executeTasksInNextMs(kArbitraryKeyframePeriodMs+1);
+}

erizoAPI/MediaStream.cc

@@ -141,6 +141,8 @@ NAN_MODULE_INIT(MediaStream::Init) {
   Nan::SetPrototypeMethod(tpl, "onMediaStreamEvent", onMediaStreamEvent);
   Nan::SetPrototypeMethod(tpl, "setVideoConstraints", setVideoConstraints);
   Nan::SetPrototypeMethod(tpl, "setMetadata", setMetadata);
+  Nan::SetPrototypeMethod(tpl, "setPeriodicKeyframeRequests", setPeriodicKeyframeRequests);
+  Nan::SetPrototypeMethod(tpl, "hasPeriodicKeyframeRequests", hasPeriodicKeyframeRequests);
   Nan::SetPrototypeMethod(tpl, "enableHandler", enableHandler);
   Nan::SetPrototypeMethod(tpl, "disableHandler", disableHandler);
 
@@ -390,6 +392,32 @@ NAN_METHOD(MediaStream::enableSlideShowBelowSpatialLayer) {
   me->enableSlideShowBelowSpatialLayer(enabled, spatial_layer);
 }
 
+NAN_METHOD(MediaStream::setPeriodicKeyframeRequests) {
+  MediaStream* obj = Nan::ObjectWrap::Unwrap<MediaStream>(info.Holder());
+  std::shared_ptr<erizo::MediaStream> me = obj->me;
+  if (!me || obj->closed_) {
+    return;
+  }
+
+  bool activated = info[0]->BooleanValue();
+  int interval = 0;
+  if (info.Length() > 1) {
+    interval = info[1]->IntegerValue();
+  }
+  me->setPeriodicKeyframeRequests(activated, interval);
+}
+
+NAN_METHOD(MediaStream::hasPeriodicKeyframeRequests) {
+  MediaStream* obj = Nan::ObjectWrap::Unwrap<MediaStream>(info.Holder());
+  std::shared_ptr<erizo::MediaStream> me = obj->me;
+  if (!me || obj->closed_) {
+    return;
+  }
+
+  bool has_periodic_requests = me->isRequestingPeriodicKeyframes();
+  info.GetReturnValue().Set(Nan::New(has_periodic_requests));
+}
+
 NAN_METHOD(MediaStream::getStats) {
   MediaStream* obj = Nan::ObjectWrap::Unwrap<MediaStream>(info.Holder());
   Nan::Callback *callback = new Nan::Callback(info[0].As<Function>());

erizoAPI/MediaStream.h

@@ -147,6 +147,8 @@ class MediaStream : public MediaSink, public erizo::MediaStreamStatsListener, pu
      * Param: An object with metadata {key1:value1, key2: value2}
      */
     static NAN_METHOD(setMetadata);
+    static NAN_METHOD(setPeriodicKeyframeRequests);
+    static NAN_METHOD(hasPeriodicKeyframeRequests);
     /*
      * Enable a specific Handler in the pipeline
      * Param: Name of the handler