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

diff --git a/modules/decklink/producer/decklink_producer.cpp b/modules/decklink/producer/decklink_producer.cpp
index 27064ceea55a72d5652f456723ad0364be46b172..600a8b2d87481a45d53a0d0a6c603587b6cacb48 100644 (file)
--- a/modules/decklink/producer/decklink_producer.cpp
+++ b/modules/decklink/producer/decklink_producer.cpp
@@ -169,6 +169,7 @@ class decklink_producer_proxy : public Concurrency::agent, public core::frame_pr
 {              \r
        Concurrency::unbounded_buffer<ffmpeg::frame_muxer2::video_source_element_t>     video_frames_;\r
        Concurrency::unbounded_buffer<ffmpeg::frame_muxer2::audio_source_element_t>     audio_buffers_;\r
+       Concurrency::overwrite_buffer<ffmpeg::frame_muxer2::target_element_t>           muxed_frames_;\r
 \r
        const core::video_format_desc           format_desc_;\r
        const size_t                                            device_index_;\r
@@ -184,11 +185,12 @@ class decklink_producer_proxy : public Concurrency::agent, public core::frame_pr
 public:\r
 \r
        explicit decklink_producer_proxy(const safe_ptr<core::frame_factory>& frame_factory, const core::video_format_desc& format_desc, size_t device_index, const std::wstring& filter_str, int64_t length)\r
-               : format_desc_(format_desc)\r
+               : muxed_frames_(1)\r
+               , format_desc_(format_desc)\r
                , device_index_(device_index)\r
                , last_frame_(core::basic_frame::empty())\r
                , length_(length)\r
-               , muxer_(&video_frames_, &audio_buffers_, format_desc.fps, frame_factory)\r
+               , muxer_(&video_frames_, &audio_buffers_, muxed_frames_, format_desc.fps, frame_factory)\r
                , is_running_(true)\r
        {\r
                agent::start();\r
@@ -206,7 +208,8 @@ public:
 \r
                try\r
                {\r
-                       last_frame_ = frame = muxer_.receive();\r
+                       auto frame_element = Concurrency::receive(muxed_frames_);\r
+                       last_frame_ = frame = frame_element.first;\r
                }\r
                catch(Concurrency::operation_timed_out&)\r
                {               \r

diff --git a/modules/ffmpeg/producer/ffmpeg_producer.cpp b/modules/ffmpeg/producer/ffmpeg_producer.cpp
index 37cf9f7049fa6553d81f6d46271168bc99966064..afb33bc95042182dd2a54a944b0678591b1adab9 100644 (file)
--- a/modules/ffmpeg/producer/ffmpeg_producer.cpp
+++ b/modules/ffmpeg/producer/ffmpeg_producer.cpp
@@ -66,6 +66,7 @@ struct ffmpeg_producer : public core::frame_producer
        unbounded_buffer<input::target_element_t>                                                               packets_;\r
        std::shared_ptr<unbounded_buffer<frame_muxer2::video_source_element_t>> video_;\r
        std::shared_ptr<unbounded_buffer<frame_muxer2::audio_source_element_t>> audio_;\r
+       unbounded_buffer<frame_muxer2::target_element_t>                                                frames_;\r
                \r
        const safe_ptr<diagnostics::graph>                                                                              graph_;\r
                                        \r
@@ -121,7 +122,7 @@ public:
                CASPAR_VERIFY(video_decoder_ || audio_decoder_, ffmpeg_error());\r
                \r
                packets_.link_target(&throw_away_);\r
-               muxer_.reset(new frame_muxer2(video_.get(), audio_.get(), video_decoder_ ? video_decoder_->fps() : frame_factory->get_video_format_desc().fps, frame_factory));\r
+               muxer_.reset(new frame_muxer2(video_.get(), audio_.get(), frames_, video_decoder_ ? video_decoder_->fps() : frame_factory->get_video_format_desc().fps, frame_factory));\r
                                \r
                graph_->set_color("underflow", diagnostics::color(0.6f, 0.3f, 0.9f));   \r
                graph_->set_text(print());\r
@@ -132,7 +133,7 @@ public:
 \r
        ~ffmpeg_producer()\r
        {\r
-               input_.stop();\r
+               input_.stop();  \r
        }\r
                                                \r
        virtual safe_ptr<core::basic_frame> receive(int hints)\r
@@ -141,7 +142,8 @@ public:
                \r
                try\r
                {               \r
-                       frame = last_frame_ = muxer_->receive();\r
+                       auto frame_element = Concurrency::receive(frames_, 10);\r
+                       frame = last_frame_ = frame_element.first;\r
                        graph_->set_text(narrow(print()));\r
                }\r
                catch(operation_timed_out&)\r

diff --git a/modules/ffmpeg/producer/frame_muxer.cpp b/modules/ffmpeg/producer/frame_muxer.cpp
index 7f89e67643e40a72d605593bf7486b457ae8c77d..9a5444ed9cb02ee1c2ff34a5a56e047b400c7d7f 100644 (file)
--- a/modules/ffmpeg/producer/frame_muxer.cpp
+++ b/modules/ffmpeg/producer/frame_muxer.cpp
@@ -39,14 +39,15 @@ using namespace Concurrency;
 \r
 namespace caspar { namespace ffmpeg {\r
        \r
-struct frame_muxer2::implementation : boost::noncopyable\r
+struct frame_muxer2::implementation : public Concurrency::agent, boost::noncopyable\r
 {              \r
-       typedef std::pair<std::shared_ptr<AVFrame>, ticket_t>           video_element_t;\r
+       typedef std::pair<safe_ptr<AVFrame>, ticket_t>                          video_element_t;\r
        typedef std::pair<safe_ptr<core::audio_buffer>, ticket_t>       audio_element_t;\r
 \r
        frame_muxer2::video_source_t* video_source_;\r
        frame_muxer2::audio_source_t* audio_source_;\r
 \r
+       ITarget<frame_muxer2::target_element_t>&                target_;\r
        mutable single_assignment<display_mode::type>   display_mode_;\r
        const double                                                                    in_fps_;\r
        const core::video_format_desc                                   format_desc_;\r
@@ -65,19 +66,27 @@ struct frame_muxer2::implementation : boost::noncopyable
        \r
        implementation(frame_muxer2::video_source_t* video_source,\r
                                   frame_muxer2::audio_source_t* audio_source,\r
+                                  frame_muxer2::target_t& target,\r
                                   double in_fps, \r
                                   const safe_ptr<core::frame_factory>& frame_factory,\r
                                   const std::wstring& filter)\r
                : video_source_(video_source)\r
                , audio_source_(audio_source)\r
+               , target_(target)\r
                , in_fps_(in_fps)\r
                , format_desc_(frame_factory->get_video_format_desc())\r
                , auto_transcode_(env::properties().get("configuration.producers.auto-transcode", false))\r
                , frame_factory_(frame_factory)\r
                , eof_(false)\r
        {               \r
+               start();\r
        }\r
-                                       \r
+\r
+       ~implementation()\r
+       {\r
+               agent::wait(this);\r
+       }\r
+                               \r
        safe_ptr<core::write_frame> receive_video(ticket_t& tickets)\r
        {       \r
                if(!video_source_)\r
@@ -88,10 +97,10 @@ struct frame_muxer2::implementation : boost::noncopyable
                        auto video_frame = std::move(video_frames_.front());\r
                        video_frames_.pop();\r
                        boost::range::push_back(tickets, video_frame.second);\r
-                       return make_write_frame(this, make_safe_ptr(video_frame.first), frame_factory_, 0);\r
+                       return make_write_frame(this, video_frame.first, frame_factory_, 0);\r
                }\r
 \r
-               auto element = Concurrency::receive(video_source_);\r
+               auto element = receive(video_source_);\r
                auto video       = element.first;\r
                \r
                if(video == loop_video())\r
@@ -107,13 +116,9 @@ struct frame_muxer2::implementation : boost::noncopyable
                        initialize_display_mode(*video);\r
                        \r
                filter_.value()->push(video);\r
-               for(auto frame = filter_.value()->poll(); frame; frame = filter_.value()->poll())               \r
-               {\r
+               for(auto frame = filter_.value()->poll(); frame; frame = filter_.value()->poll())                       \r
                        video_frames_.push(video_element_t(make_safe_ptr(frame), element.second));              \r
-                       if(display_mode_.value() == display_mode::duplicate)\r
-                               video_frames_.push(video_element_t(make_safe_ptr(frame), element.second));      \r
-               }\r
-\r
+               \r
                return receive_video(tickets);\r
        }\r
        \r
@@ -130,7 +135,7 @@ struct frame_muxer2::implementation : boost::noncopyable
                        return audio_buffer.first;\r
                }\r
                \r
-               auto element = Concurrency::receive(audio_source_);\r
+               auto element = receive(audio_source_);\r
                auto audio       = element.first;\r
 \r
                if(audio == loop_audio())\r
@@ -162,51 +167,70 @@ struct frame_muxer2::implementation : boost::noncopyable
                return receive_audio(tickets);\r
        }\r
                        \r
-       safe_ptr<core::basic_frame> receive_frame()\r
+       virtual void run()\r
        {\r
                try\r
                {\r
-                       if(eof_)\r
-                               return core::basic_frame::eof();\r
-\r
-                       ticket_t tickets;\r
+                       while(!eof_)\r
+                       {\r
+                               ticket_t tickets;\r
                                \r
-                       auto video                      = receive_video(tickets);\r
-                       video->audio_data() = std::move(*receive_audio(tickets));\r
+                               auto video = receive_video(tickets);\r
+                               video->audio_data() = std::move(*receive_audio(tickets));\r
 \r
-                       if(eof_)\r
-                               return core::basic_frame::eof();\r
+                               if(eof_)\r
+                                       break;\r
 \r
-                       switch(display_mode_.value())\r
-                       {\r
-                       case display_mode::simple:                      \r
-                       case display_mode::deinterlace:\r
-                       case display_mode::deinterlace_bob:\r
-                       case display_mode::duplicate:\r
+                               switch(display_mode_.value())\r
                                {\r
-                                       return video;\r
-                               }               \r
-                       case display_mode::half:                                                \r
-                               {                               \r
-                                       receive_video(tickets);                         \r
-                                       return video;\r
-                               }\r
-                       case display_mode::deinterlace_bob_reinterlace:\r
-                       case display_mode::interlace:                                   \r
-                               {                                       \r
-                                       auto video2 = receive_video(tickets);                                                                                           \r
-                                       return core::basic_frame::interlace(std::move(video), std::move(video2), format_desc_.field_mode);      \r
+                               case display_mode::simple:                      \r
+                               case display_mode::deinterlace:\r
+                               case display_mode::deinterlace_bob:\r
+                                       {\r
+                                               send(target_, frame_muxer2::target_element_t(std::move(video), std::move(tickets)));\r
+\r
+                                               break;\r
+                                       }\r
+                               case display_mode::duplicate:                                   \r
+                                       {                                               \r
+                                               auto video2 = make_safe<core::write_frame>(*video);     \r
+                                               video2->audio_data() = std::move(*receive_audio(tickets));\r
+\r
+                                               send(target_, frame_muxer2::target_element_t(std::move(video), std::move(tickets)));                                            \r
+                                               send(target_, frame_muxer2::target_element_t(std::move(video2), std::move(tickets)));\r
+\r
+                                               break;\r
+                                       }\r
+                               case display_mode::half:                                                \r
+                                       {                                                               \r
+                                               send(target_, frame_muxer2::target_element_t(std::move(video), std::move(tickets)));\r
+                                               receive_video(tickets);\r
+\r
+                                               break;\r
+                                       }\r
+                               case display_mode::deinterlace_bob_reinterlace:\r
+                               case display_mode::interlace:                                   \r
+                                       {                                       \r
+                                               auto video2 = receive_video(tickets);\r
+                                                                                               \r
+                                               auto frame = core::basic_frame::interlace(std::move(video), std::move(video2), format_desc_.field_mode);        \r
+                                               send(target_, frame_muxer2::target_element_t(std::move(frame), std::move(tickets)));\r
+\r
+                                               break;\r
+                                       }\r
+                               default:        \r
+                                       BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("invalid display-mode"));\r
                                }\r
-                       default:        \r
-                               BOOST_THROW_EXCEPTION(invalid_operation() << msg_info("invalid display-mode"));\r
-                       }\r
+                       }       \r
                }\r
                catch(...)\r
                {\r
                        CASPAR_LOG_CURRENT_EXCEPTION();\r
-                       eof_ = true;\r
-                       return core::basic_frame::eof();\r
                }\r
+               \r
+               send(target_, frame_muxer2::target_element_t(core::basic_frame::eof(), ticket_t()));\r
+\r
+               done();\r
        }\r
 \r
        void initialize_display_mode(AVFrame& frame)\r
@@ -273,10 +297,11 @@ struct frame_muxer2::implementation : boost::noncopyable
 \r
 frame_muxer2::frame_muxer2(video_source_t* video_source, \r
                                                   audio_source_t* audio_source,\r
+                                                  target_t& target,\r
                                                   double in_fps, \r
                                                   const safe_ptr<core::frame_factory>& frame_factory,\r
                                                   const std::wstring& filter)\r
-       : impl_(new implementation(video_source, audio_source, in_fps, frame_factory, filter))\r
+       : impl_(new implementation(video_source, audio_source, target, in_fps, frame_factory, filter))\r
 {\r
 }\r
 \r
@@ -285,9 +310,4 @@ int64_t frame_muxer2::calc_nb_frames(int64_t nb_frames) const
        return impl_->calc_nb_frames(nb_frames);\r
 }\r
 \r
-safe_ptr<core::basic_frame> frame_muxer2::receive()\r
-{\r
-       return impl_->receive_frame();\r
-}\r
-\r
 }}
\ No newline at end of file

diff --git a/modules/ffmpeg/producer/frame_muxer.h b/modules/ffmpeg/producer/frame_muxer.h
index 48f5e8db80280f9bc3d65cfca701b5f1a7c01b49..a2940c6974d8f2c28ba661927f2475ae08dd1f5f 100644 (file)
--- a/modules/ffmpeg/producer/frame_muxer.h
+++ b/modules/ffmpeg/producer/frame_muxer.h
@@ -33,18 +33,19 @@ public:
        \r
        typedef std::pair<safe_ptr<AVFrame>, ticket_t>                          video_source_element_t;\r
        typedef std::pair<safe_ptr<core::audio_buffer>, ticket_t>       audio_source_element_t;\r
+       typedef std::pair<safe_ptr<core::basic_frame>, ticket_t>        target_element_t;\r
 \r
        typedef Concurrency::ISource<video_source_element_t>            video_source_t;\r
        typedef Concurrency::ISource<audio_source_element_t>            audio_source_t;\r
+       typedef Concurrency::ITarget<target_element_t>                          target_t;\r
                                                                 \r
        frame_muxer2(video_source_t* video_source,\r
                                 audio_source_t* audio_source, \r
+                                target_t& target,\r
                                 double in_fps, \r
                                 const safe_ptr<core::frame_factory>& frame_factory,\r
                                 const std::wstring& filter = L"");\r
        \r
-       safe_ptr<core::basic_frame> receive();\r
-\r
        int64_t calc_nb_frames(int64_t nb_frames) const;\r
 private:\r
        struct implementation;\r