Refine JobTimer and add SharedJobTimer (#606)

Author: starwarfan

source/core/common/JobTimer.cpp

@@ -6,6 +6,7 @@
 
 #include <boost/thread.hpp>
 #include <mutex>
+#include <unordered_map>
 
 namespace {
 
@@ -89,9 +90,9 @@ void JobTimer::onTimeout(const boost::system::error_code& ec)
 {
     if (!ec) {
         if (!m_isClosing) {
-            handleJob();
             m_timer->expires_from_now(boost::posix_time::milliseconds(m_interval));
             m_timer->async_wait(boost::bind(&JobTimer::onTimeout, this, boost::asio::placeholders::error));
+            handleJob();
         }
     }
 }
@@ -101,3 +102,53 @@ void JobTimer::handleJob()
     if (m_listener)
         m_listener->onTimeout();
 }
+
+SharedJobTimer::SharedJobTimer(unsigned int frequency)
+    : m_jobTimer(frequency, this)
+{
+    m_jobTimer.start();
+}
+
+SharedJobTimer::~SharedJobTimer()
+{
+    m_jobTimer.stop();
+}
+
+void SharedJobTimer::addListener(JobTimerListener* listener)
+{
+    if (listener) {
+        boost::mutex::scoped_lock lock(m_mutex);
+        m_listeners.insert(listener);
+    }
+}
+
+void SharedJobTimer::removeListener(JobTimerListener* listener)
+{
+    if (listener) {
+        boost::mutex::scoped_lock lock(m_mutex);
+        m_listeners.erase(listener);
+    }
+}
+
+void SharedJobTimer::onTimeout()
+{
+    boost::mutex::scoped_lock lock(m_mutex);
+    for (JobTimerListener* listener : m_listeners) {
+        listener->onTimeout();
+    }
+}
+
+std::shared_ptr<SharedJobTimer> SharedJobTimer::GetSharedFrequencyTimer(unsigned int frequency)
+{
+    static boost::mutex timersMutex;
+    static std::unordered_map<unsigned int, std::shared_ptr<SharedJobTimer>> sharedTimers;
+
+    boost::mutex::scoped_lock lock(timersMutex);
+    if (sharedTimers.count(frequency) > 0) {
+        return sharedTimers[frequency];
+    } else {
+        auto sharedTimer = std::make_shared<SharedJobTimer>(frequency);
+        sharedTimers.emplace(frequency, sharedTimer);
+        return sharedTimer;
+    }
+}

source/core/common/JobTimer.h

@@ -36,4 +36,21 @@ class JobTimer {
     boost::scoped_ptr<boost::asio::deadline_timer> m_timer;
 };
 
+class SharedJobTimer : public JobTimerListener {
+public:
+    SharedJobTimer(unsigned int frequency);
+    ~SharedJobTimer();
+
+    static std::shared_ptr<SharedJobTimer> GetSharedFrequencyTimer(unsigned int frequency);
+    void addListener(JobTimerListener* listener);
+    void removeListener(JobTimerListener* listener);
+
+    virtual void onTimeout() override;
+
+private:
+    JobTimer m_jobTimer;
+    boost::mutex m_mutex;
+    std::set<JobTimerListener*> m_listeners;
+};
+
 #endif

source/core/owt_base/MediaFrameMulticaster.cpp

@@ -9,12 +9,13 @@ namespace owt_base {
 MediaFrameMulticaster::MediaFrameMulticaster()
     : m_pendingKeyFrameRequests(0)
 {
-    m_feedbackTimer.reset(new JobTimer(1, this));
+    m_feedbackTimer = SharedJobTimer::GetSharedFrequencyTimer(1);
+    m_feedbackTimer->addListener(this);
 }
 
 MediaFrameMulticaster::~MediaFrameMulticaster()
 {
-    m_feedbackTimer->stop();
+    m_feedbackTimer->removeListener(this);
 }
 
 void MediaFrameMulticaster::onFeedback(const FeedbackMsg& msg)

source/core/owt_base/MediaFrameMulticaster.h

@@ -25,7 +25,7 @@ class MediaFrameMulticaster : public FrameSource, public FrameDestination, publi
     void onTimeout();
 
 private:
-    boost::scoped_ptr<JobTimer> m_feedbackTimer;
+    std::shared_ptr<SharedJobTimer> m_feedbackTimer;
     uint32_t m_pendingKeyFrameRequests;
 };
 

source/core/owt_base/VideoFrameConstructor.cpp

@@ -24,7 +24,8 @@ VideoFrameConstructor::VideoFrameConstructor(VideoInfoListener* vil, uint32_t tr
     , m_videoReceive(nullptr)
 {
     m_config.transport_cc = transportccExtId;
-    m_feedbackTimer.reset(new JobTimer(1, this));
+    m_feedbackTimer = SharedJobTimer::GetSharedFrequencyTimer(1);
+    m_feedbackTimer->addListener(this);
 }
 
 VideoFrameConstructor::VideoFrameConstructor(
@@ -39,13 +40,14 @@ VideoFrameConstructor::VideoFrameConstructor(
 {
     m_config.transport_cc = transportccExtId;
     assert(base);
-    m_feedbackTimer.reset(new JobTimer(1, this));
+    m_feedbackTimer = SharedJobTimer::GetSharedFrequencyTimer(1);
+    m_feedbackTimer->addListener(this);
     m_rtcAdapter = base->m_rtcAdapter;
 }
 
 VideoFrameConstructor::~VideoFrameConstructor()
 {
-    m_feedbackTimer->stop();
+    m_feedbackTimer->removeListener(this);
     unbindTransport();
     if (m_videoReceive) {
         m_rtcAdapter->destoryVideoReceiver(m_videoReceive);

source/core/owt_base/VideoFrameConstructor.h

@@ -83,7 +83,7 @@ class VideoFrameConstructor : public erizo::MediaSink,
 
     erizo::MediaSource* m_transport;
     boost::shared_mutex m_transportMutex;
-    boost::scoped_ptr<JobTimer> m_feedbackTimer;
+    std::shared_ptr<SharedJobTimer> m_feedbackTimer;
     uint32_t m_pendingKeyFrameRequests;
 
     VideoInfoListener* m_videoInfoListener;