* Merged streaming_consumer from 2.0

[casparcg] / core / consumer / frame_consumer.cpp

diff --git a/core/consumer/frame_consumer.cpp b/core/consumer/frame_consumer.cpp
index 3cc4e768ba5910812a833b9ddd36e648e348da0e..b695f5a84fb58d01434cb005fe0e59ec86b26906 100644 (file)
--- a/core/consumer/frame_consumer.cpp
+++ b/core/consumer/frame_consumer.cpp
@@ -32,14 +32,24 @@
 #include <boost/thread.hpp>
 
 #include <future>
+#include <vector>
+#include <map>
 
 namespace caspar { namespace core {
                
-std::vector<consumer_factory_t> g_factories;
+std::vector<consumer_factory_t> g_consumer_factories;
+std::map<std::wstring, preconfigured_consumer_factory_t> g_preconfigured_consumer_factories;
 
 void register_consumer_factory(const consumer_factory_t& factory)
 {
-       g_factories.push_back(factory);
+       g_consumer_factories.push_back(factory);
+}
+
+void register_preconfigured_consumer_factory(
+               const std::wstring& element_name,
+               const preconfigured_consumer_factory_t& factory)
+{
+       g_preconfigured_consumer_factories.insert(std::make_pair(element_name, factory));
 }
 
 class destroy_consumer_proxy : public frame_consumer
@@ -53,7 +63,9 @@ public:
 
        ~destroy_consumer_proxy()
        {               
-               static tbb::atomic<int> counter = tbb::atomic<int>();
+               static tbb::atomic<int> counter;
+               static std::once_flag counter_init_once;
+               std::call_once(counter_init_once, []{ counter = 0; });
                        
                ++counter;
                CASPAR_VERIFY(counter < 8);
@@ -75,12 +87,11 @@ public:
 
                        pointer_guard.reset();
 
-                       --counter;
                }).detach(); 
        }
        
        std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
-       virtual void initialize(const struct video_format_desc& format_desc, int channel_index) override        {return consumer_->initialize(format_desc, channel_index);}
+       virtual void initialize(const video_format_desc& format_desc, int channel_index)        override        {return consumer_->initialize(format_desc, channel_index);}
        std::wstring print() const override                                                                                                                                     {return consumer_->print();}    
        std::wstring name() const override                                                                                                                                      {return consumer_->name();}
        boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
@@ -109,7 +120,7 @@ public:
        }
        
        std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
-       virtual void initialize(const struct video_format_desc& format_desc, int channel_index) override        {return consumer_->initialize(format_desc, channel_index);}
+       virtual void initialize(const video_format_desc& format_desc, int channel_index)        override        {return consumer_->initialize(format_desc, channel_index);}
        std::wstring print() const override                                                                                                                                     {return consumer_->print();}
        std::wstring name() const override                                                                                                                                      {return consumer_->name();}
        boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
@@ -153,7 +164,7 @@ public:
                }
        }
 
-       virtual void initialize(const struct video_format_desc& format_desc, int channel_index)         
+       virtual void initialize(const video_format_desc& format_desc, int channel_index)                
        {
                format_desc_    = format_desc;
                channel_index_  = channel_index;
@@ -220,17 +231,18 @@ public:
        monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}                                                                           
 };
 
-spl::shared_ptr<core::frame_consumer> create_consumer(const std::vector<std::wstring>& params)
+spl::shared_ptr<core::frame_consumer> create_consumer(
+               const std::vector<std::wstring>& params, interaction_sink* sink)
 {
        if(params.empty())
                CASPAR_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
        
        auto consumer = frame_consumer::empty();
-       std::any_of(g_factories.begin(), g_factories.end(), [&](const consumer_factory_t& factory) -> bool
+       std::any_of(g_consumer_factories.begin(), g_consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
                {
                        try
                        {
-                               consumer = factory(params);
+                               consumer = factory(params, sink);
                        }
                        catch(...)
                        {
@@ -249,6 +261,24 @@ spl::shared_ptr<core::frame_consumer> create_consumer(const std::vector<std::wst
                           std::move(consumer)))));
 }
 
+spl::shared_ptr<frame_consumer> create_consumer(
+               const std::wstring& element_name,
+               const boost::property_tree::wptree& element,
+               interaction_sink* sink)
+{
+       auto found = g_preconfigured_consumer_factories.find(element_name);
+
+       if (found == g_preconfigured_consumer_factories.end())
+               CASPAR_THROW_EXCEPTION(file_not_found()
+                       << msg_info(L"No consumer factory registered for element name " + element_name));
+
+       return spl::make_shared<destroy_consumer_proxy>(
+                       spl::make_shared<print_consumer_proxy>(
+                                       spl::make_shared<recover_consumer_proxy>(
+                                                       spl::make_shared<cadence_guard>(
+                                                                       found->second(element, sink)))));
+}
+
 const spl::shared_ptr<frame_consumer>& frame_consumer::empty()
 {
        class empty_frame_consumer : public frame_consumer