Created a consumer that provides sync to a channel based on the pace of another chann...

[casparcg] / core / video_channel.cpp

diff --git a/core/video_channel.cpp b/core/video_channel.cpp
index 78520b060bf8482dba2563acc5a636ab85f976fb..a0ac2c2012fd249ac76d0fcd85a8483826ee7e86 100644 (file)
--- a/core/video_channel.cpp
+++ b/core/video_channel.cpp
@@ -38,6 +38,7 @@
 #include <common/lock.h>
 #include <common/executor.h>
 #include <common/timer.h>
+#include <common/future.h>
 
 #include <core/mixer/image/image_mixer.h>
 #include <core/diagnostics/call_context.h>
@@ -48,6 +49,7 @@
 #include <boost/lexical_cast.hpp>
 
 #include <string>
+#include <unordered_map>
 
 namespace caspar { namespace core {
 
@@ -62,19 +64,24 @@ struct video_channel::impl final
        mutable tbb::spin_mutex                                                         channel_layout_mutex_;
        core::audio_channel_layout                                                      channel_layout_;
 
-       const spl::shared_ptr<caspar::diagnostics::graph>       graph_                          = [](int index)
-                                                                                                                                                         {
-                                                                                                                                                                 core::diagnostics::scoped_call_context save;
-                                                                                                                                                                 core::diagnostics::call_context::for_thread().video_channel = index;
-                                                                                                                                                                 return spl::make_shared<caspar::diagnostics::graph>();
-                                                                                                                                                         }(index_);
+       const spl::shared_ptr<caspar::diagnostics::graph>       graph_                                  = [](int index)
+                                                                                                                                                                 {
+                                                                                                                                                                         core::diagnostics::scoped_call_context save;
+                                                                                                                                                                         core::diagnostics::call_context::for_thread().video_channel = index;
+                                                                                                                                                                         return spl::make_shared<caspar::diagnostics::graph>();
+                                                                                                                                                                 }(index_);
 
        caspar::core::output                                                            output_;
+       std::future<void>                                                                       output_ready_for_frame_ = make_ready_future();
        spl::shared_ptr<image_mixer>                                            image_mixer_;
        caspar::core::mixer                                                                     mixer_;
-       caspar::core::stage                                                                     stage_; 
+       caspar::core::stage                                                                     stage_;
 
-       executor                                                                                        executor_                       { L"video_channel " + boost::lexical_cast<std::wstring>(index_) };
+       mutable tbb::spin_mutex                                                         tick_listeners_mutex_;
+       int64_t                                                                                         last_tick_listener_id   = 0;
+       std::unordered_map<int64_t, std::function<void ()>>     tick_listeners_;
+
+       executor                                                                                        executor_                               { L"video_channel " + boost::lexical_cast<std::wstring>(index_) };
 public:
        impl(
                        int index,
@@ -143,32 +150,51 @@ public:
                });
        }
 
+       void invoke_tick_listeners()
+       {
+               auto listeners = lock(tick_listeners_mutex_, [=] { return tick_listeners_; });
+
+               for (auto listener : listeners)
+               {
+                       try
+                       {
+                               listener.second();
+                       }
+                       catch (...)
+                       {
+                               CASPAR_LOG_CURRENT_EXCEPTION();
+                       }
+               }
+       }
+
        void tick()
        {
                try
                {
+                       invoke_tick_listeners();
 
                        auto format_desc        = video_format_desc();
                        auto channel_layout = audio_channel_layout();
-                       
+
                        caspar::timer frame_timer;
 
                        // Produce
-                       
+
                        auto stage_frames = stage_(format_desc);
-                       
+
                        // Mix
-                       
+
                        auto mixed_frame  = mixer_(std::move(stage_frames), format_desc, channel_layout);
-                       
+
                        // Consume
-                                               
-                       output_(std::move(mixed_frame), format_desc, channel_layout);
-               
+
+                       output_ready_for_frame_ = output_(std::move(mixed_frame), format_desc, channel_layout);
+                       output_ready_for_frame_.get();
+
                        auto frame_time = frame_timer.elapsed()*format_desc.fps*0.5;
                        graph_->set_value("tick-time", frame_time);
 
-                       *monitor_subject_       << monitor::message("/profiler/time")   % frame_timer.elapsed() % (1.0/format_desc_.fps)
+                       *monitor_subject_       << monitor::message("/profiler/time")   % frame_timer.elapsed() % (1.0/ video_format_desc().fps)
                                                                << monitor::message("/format")                  % format_desc.name;
                }
                catch(...)
@@ -179,7 +205,7 @@ public:
                if (executor_.is_running())
                        executor_.begin_invoke([=]{tick();});
        }
-                       
+
        std::wstring print() const
        {
                return L"video_channel[" + boost::lexical_cast<std::wstring>(index_) + L"|" +  video_format_desc().name + L"]";
@@ -203,8 +229,8 @@ public:
                info.add_child(L"stage", stage_info.get());
                info.add_child(L"mixer", mixer_info.get());
                info.add_child(L"output", output_info.get());
-   
-               return info;                       
+
+               return info;
        }
 
        boost::property_tree::wptree delay_info() const
@@ -222,6 +248,23 @@ public:
 
                return info;
        }
+
+       std::shared_ptr<void> add_tick_listener(std::function<void()> listener)
+       {
+               return lock(tick_listeners_mutex_, [&]
+               {
+                       auto tick_listener_id = last_tick_listener_id++;
+                       tick_listeners_.insert(std::make_pair(tick_listener_id, listener));
+
+                       return std::shared_ptr<void>(nullptr, [=](void*)
+                       {
+                               lock(tick_listeners_mutex_, [&]
+                               {
+                                       tick_listeners_.erase(tick_listener_id);
+                               });
+                       });
+               });
+       }
 };
 
 video_channel::video_channel(
@@ -245,5 +288,6 @@ boost::property_tree::wptree video_channel::info() const{return impl_->info();}
 boost::property_tree::wptree video_channel::delay_info() const { return impl_->delay_info(); }
 int video_channel::index() const { return impl_->index(); }
 monitor::subject& video_channel::monitor_output(){ return *impl_->monitor_subject_; }
+std::shared_ptr<void> video_channel::add_tick_listener(std::function<void()> listener) { return impl_->add_tick_listener(std::move(listener)); }
 
 }}