git-svn-id: https://casparcg.svn.sourceforge.net/svnroot/casparcg/server/branches...

diff --git a/core/consumer/frame_consumer.cpp b/core/consumer/frame_consumer.cpp
index 02ed56b8535251819ac22598c9adb9e4f37006d8..b6c5d38a4faa77655241b32ca575f4c5005108b8 100644 (file)
--- a/core/consumer/frame_consumer.cpp
+++ b/core/consumer/frame_consumer.cpp
@@ -25,18 +25,89 @@
 #include <common/memory/safe_ptr.h>\r
 #include <common/exception/exceptions.h>\r
 \r
-namespace caspar { namespace core {\r
+#include <concurrent_vector.h>\r
 \r
+namespace caspar { namespace core {\r
+       \r
 size_t consumer_buffer_depth()\r
 {\r
        return env::properties().get("configuration.consumers.buffer-depth", 5);\r
 }\r
+\r
+struct destruction_context\r
+{\r
+       std::shared_ptr<frame_consumer> consumer;\r
+       Concurrency::event                              event;\r
+\r
+       destruction_context(std::shared_ptr<frame_consumer>&& consumer) \r
+               : consumer(consumer)\r
+       {\r
+       }\r
+};\r
+\r
+void __cdecl destroy_consumer(LPVOID lpParam)\r
+{\r
+       auto destruction = std::unique_ptr<destruction_context>(static_cast<destruction_context*>(lpParam));\r
+       \r
+       try\r
+       {               \r
+               if(destruction->consumer.unique())\r
+               {\r
+                       Concurrency::scoped_oversubcription_token oversubscribe;\r
+                       destruction->consumer.reset();\r
+               }\r
+               else\r
+                       CASPAR_LOG(warning) << destruction->consumer->print() << " Not destroyed asynchronously.";              \r
+       }\r
+       catch(...)\r
+       {\r
+               CASPAR_LOG_CURRENT_EXCEPTION();\r
+       }\r
        \r
-std::vector<const consumer_factory_t> g_factories;\r
+       destruction->event.set();\r
+}\r
+\r
+void __cdecl destroy_and_wait_consumer(LPVOID lpParam)\r
+{\r
+       try\r
+       {\r
+               auto destruction = static_cast<destruction_context*>(lpParam);\r
+               Concurrency::CurrentScheduler::ScheduleTask(destroy_consumer, lpParam);\r
+               if(destruction->event.wait(1000) == Concurrency::COOPERATIVE_WAIT_TIMEOUT)\r
+                       CASPAR_LOG(warning) << " Potential destruction deadlock detected. Might leak resources.";\r
+       }\r
+       catch(...)\r
+       {\r
+               CASPAR_LOG_CURRENT_EXCEPTION();\r
+       }\r
+}\r
+       \r
+class destroy_consumer_proxy : public frame_consumer\r
+{\r
+       std::shared_ptr<frame_consumer> consumer_;\r
+public:\r
+       destroy_consumer_proxy(const std::shared_ptr<frame_consumer>& consumer) \r
+               : consumer_(consumer)\r
+       {\r
+       }\r
+\r
+       ~destroy_consumer_proxy()\r
+       {               \r
+               Concurrency::CurrentScheduler::ScheduleTask(destroy_consumer, new destruction_context(std::move(consumer_)));\r
+       }       \r
+       \r
+       virtual bool send(const safe_ptr<read_frame>& frame)                                    {return consumer_->send(frame);}\r
+       virtual void initialize(const video_format_desc& format_desc)                   {return consumer_->initialize(format_desc);}\r
+       virtual std::wstring print() const                                                                              {return consumer_->print();}\r
+       virtual bool has_synchronization_clock() const                                                  {return consumer_->has_synchronization_clock();}\r
+       virtual const core::video_format_desc& get_video_format_desc() const    {return consumer_->get_video_format_desc();}\r
+};\r
+\r
+Concurrency::concurrent_vector<std::shared_ptr<consumer_factory_t>> g_factories;\r
 \r
 void register_consumer_factory(const consumer_factory_t& factory)\r
 {\r
-       g_factories.push_back(factory);\r
+       g_factories.push_back(std::make_shared<consumer_factory_t>(factory));\r
 }\r
 \r
 safe_ptr<core::frame_consumer> create_consumer(const std::vector<std::wstring>& params)\r
@@ -45,11 +116,11 @@ safe_ptr<core::frame_consumer> create_consumer(const std::vector<std::wstring>&
                BOOST_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));\r
        \r
        auto consumer = frame_consumer::empty();\r
-       std::any_of(g_factories.begin(), g_factories.end(), [&](const consumer_factory_t& factory) -> bool\r
+       std::any_of(g_factories.begin(), g_factories.end(), [&](const std::shared_ptr<consumer_factory_t>& factory) -> bool\r
                {\r
                        try\r
                        {\r
-                               consumer = factory(params);\r
+                               consumer = (*factory)(params);\r
                        }\r
                        catch(...)\r
                        {\r
@@ -61,7 +132,7 @@ safe_ptr<core::frame_consumer> create_consumer(const std::vector<std::wstring>&
        if(consumer == frame_consumer::empty())\r
                BOOST_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax."));\r
 \r
-       return consumer;\r
+       return make_safe<destroy_consumer_proxy>(consumer);\r
 }\r
 \r
 }}
\ No newline at end of file

diff --git a/core/consumer/output.cpp b/core/consumer/output.cpp
index e8adaca5e5d6dde92e1db5af56a29e913edc0da8..c863ee6200a0a199d7f5c4dfe4500539e628707b 100644 (file)
--- a/core/consumer/output.cpp
+++ b/core/consumer/output.cpp
@@ -41,54 +41,6 @@ using namespace Concurrency;
 \r
 namespace caspar { namespace core {\r
        \r
-struct destruction_context\r
-{\r
-       std::shared_ptr<frame_consumer> consumer;\r
-       Concurrency::event                              event;\r
-\r
-       destruction_context(std::shared_ptr<frame_consumer>&& consumer) \r
-               : consumer(consumer)\r
-       {\r
-       }\r
-};\r
-\r
-void __cdecl destroy_consumer(LPVOID lpParam)\r
-{\r
-       auto destruction = std::unique_ptr<destruction_context>(static_cast<destruction_context*>(lpParam));\r
-       \r
-       try\r
-       {               \r
-               if(destruction->consumer.unique())\r
-               {\r
-                       Concurrency::scoped_oversubcription_token oversubscribe;\r
-                       destruction->consumer.reset();\r
-               }\r
-               else\r
-                       CASPAR_LOG(warning) << destruction->consumer->print() << " Not destroyed asynchronously.";              \r
-       }\r
-       catch(...)\r
-       {\r
-               CASPAR_LOG_CURRENT_EXCEPTION();\r
-       }\r
-       \r
-       destruction->event.set();\r
-}\r
-\r
-void __cdecl destroy_and_wait_consumer(LPVOID lpParam)\r
-{\r
-       try\r
-       {\r
-               auto destruction = static_cast<destruction_context*>(lpParam);\r
-               Concurrency::CurrentScheduler::ScheduleTask(destroy_consumer, lpParam);\r
-               if(destruction->event.wait(1000) == Concurrency::COOPERATIVE_WAIT_TIMEOUT)\r
-                       CASPAR_LOG(warning) << " Potential destruction deadlock detected. Might leak resources.";\r
-       }\r
-       catch(...)\r
-       {\r
-               CASPAR_LOG_CURRENT_EXCEPTION();\r
-       }\r
-}\r
-\r
 struct output::implementation\r
 {      \r
        typedef std::pair<safe_ptr<read_frame>, safe_ptr<read_frame>> fill_and_key;\r
@@ -178,11 +130,7 @@ public:
                });\r
 \r
                BOOST_FOREACH(auto& removable, removables)\r
-               {\r
-                       std::shared_ptr<frame_consumer> consumer = consumers_.find(removable)->second;\r
                        consumers_.erase(removable);                    \r
-                       Concurrency::CurrentScheduler::ScheduleTask(destroy_consumer, new destruction_context(std::move(consumer)));\r
-               }\r
        }\r
 \r
 private:\r

diff --git a/core/producer/frame_producer.cpp b/core/producer/frame_producer.cpp
index 50e61b328338df93ca00f726ab4b2e6e0e221a8a..1d4e6d133d520b797598745a3e74991dd5fc78c7 100644 (file)
--- a/core/producer/frame_producer.cpp
+++ b/core/producer/frame_producer.cpp
@@ -31,11 +31,10 @@
 #include <common/exception/exceptions.h>\r
 \r
 #include <concrt_extras.h>\r
+#include <concurrent_vector.h>\r
 \r
 namespace caspar { namespace core {\r
        \r
-std::vector<const producer_factory_t> g_factories;\r
-       \r
 struct destruction_context\r
 {\r
        std::shared_ptr<frame_producer> producer;\r
@@ -107,11 +106,6 @@ public:
        virtual int64_t                                         nb_frames() const                                                                                               {return producer_->nb_frames();}\r
 };\r
 \r
-safe_ptr<core::frame_producer> create_destroy_producer_proxy(const safe_ptr<frame_producer>& producer)\r
-{\r
-       return make_safe<destroy_producer_proxy>(producer);\r
-}\r
-\r
 class last_frame_producer : public frame_producer\r
 {\r
        const std::wstring                      print_;\r
@@ -165,10 +159,12 @@ safe_ptr<basic_frame> receive_and_follow(safe_ptr<frame_producer>& producer, int
        }\r
        return frame;\r
 }\r
+       \r
+Concurrency::concurrent_vector<std::shared_ptr<producer_factory_t>> g_factories;\r
 \r
 void register_producer_factory(const producer_factory_t& factory)\r
 {\r
-       g_factories.push_back(factory);\r
+       g_factories.push_back(std::make_shared<producer_factory_t>(factory));\r
 }\r
 \r
 safe_ptr<core::frame_producer> do_create_producer(const safe_ptr<frame_factory>& my_frame_factory, const std::vector<std::wstring>& params)\r
@@ -177,11 +173,11 @@ safe_ptr<core::frame_producer> do_create_producer(const safe_ptr<frame_factory>&
                BOOST_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));\r
        \r
        auto producer = frame_producer::empty();\r
-       std::any_of(g_factories.begin(), g_factories.end(), [&](const producer_factory_t& factory) -> bool\r
+       std::any_of(g_factories.begin(), g_factories.end(), [&](const std::shared_ptr<producer_factory_t>& factory) -> bool\r
                {\r
                        try\r
                        {\r
-                               producer = factory(my_frame_factory, params);\r
+                               producer = (*factory)(my_frame_factory, params);\r
                        }\r
                        catch(...)\r
                        {\r
@@ -193,10 +189,9 @@ safe_ptr<core::frame_producer> do_create_producer(const safe_ptr<frame_factory>&
        if(producer == frame_producer::empty())\r
                producer = create_color_producer(my_frame_factory, params);\r
        \r
-       return producer;\r
+       return make_safe<destroy_producer_proxy>(producer);\r
 }\r
 \r
-\r
 safe_ptr<core::frame_producer> create_producer(const safe_ptr<frame_factory>& my_frame_factory, const std::vector<std::wstring>& params)\r
 {      \r
        auto producer = do_create_producer(my_frame_factory, params);\r

diff --git a/core/producer/frame_producer.h b/core/producer/frame_producer.h
index 067a4ee5f9cfe490cc75a94102d8b965b5428537..c19fec2118cc0df1ffcf3f5635c85c0f202cb414 100644 (file)
--- a/core/producer/frame_producer.h
+++ b/core/producer/frame_producer.h
@@ -71,8 +71,6 @@ typedef std::function<safe_ptr<core::frame_producer>(const safe_ptr<frame_factor
 void register_producer_factory(const producer_factory_t& factory); // Not thread-safe.\r
 safe_ptr<core::frame_producer> create_producer(const safe_ptr<frame_factory>&, const std::vector<std::wstring>& params);\r
 \r
-safe_ptr<core::frame_producer> create_destroy_producer_proxy(const safe_ptr<frame_producer>& producer);\r
-\r
 template<typename T>\r
 typename std::decay<T>::type get_param(const std::wstring& name, const std::vector<std::wstring>& params, T fail_value)\r
 {      \r

diff --git a/core/producer/stage.cpp b/core/producer/stage.cpp
index ff9d4d26e9cd415b1c1c1533cc5b13ef711c9118..a86ec13af9107bc16ee09459761a7b19d4373dd9 100644 (file)
--- a/core/producer/stage.cpp
+++ b/core/producer/stage.cpp
@@ -100,7 +100,7 @@ public:
        void load(int index, const safe_ptr<frame_producer>& producer, bool preview, int auto_play_delta)\r
        {\r
                critical_section::scoped_lock lock(mutex_);\r
-               layers_[index].load(create_destroy_producer_proxy(producer), preview, auto_play_delta);\r
+               layers_[index].load(producer, preview, auto_play_delta);\r
        }\r
 \r
        void pause(int index)\r