git.sesse.net Git - casparcg/blob - modules/reroute/producer/reroute_producer.cpp

   1 /*\r
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>\r
   3 *\r
   4 * This file is part of CasparCG (www.casparcg.com).\r
   5 *\r
   6 * CasparCG is free software: you can redistribute it and/or modify\r
   7 * it under the terms of the GNU General Public License as published by\r
   8 * the Free Software Foundation, either version 3 of the License, or\r
   9 * (at your option) any later version.\r
  10 *\r
  11 * CasparCG is distributed in the hope that it will be useful,\r
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of\r
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\r
  14 * GNU General Public License for more details.\r
  15 *\r
  16 * You should have received a copy of the GNU General Public License\r
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.\r
  18 *\r
  19 * Author: Robert Nagy, ronag89@gmail.com\r
  20 */\r
  21 \r
  22 #include "../stdafx.h"\r
  23 \r
  24 #include "reroute_producer.h"\r
  25 \r
  26 #include <core/producer/frame_producer.h>\r
  27 #include <core/frame/draw_frame.h>\r
  28 #include <core/frame/frame_factory.h>\r
  29 #include <core/frame/pixel_format.h>\r
  30 #include <core/mixer/gpu/write_frame.h>\r
  31 #include <core/frame/data_frame.h>\r
  32 \r
  33 #include <common/exception/exceptions.h>\r
  34 #include <common/diagnostics/graph.h>\r
  35 #include <common/log.h>\r
  36 #include <common/reactive.h>\r
  37 \r
  38 #include <asmlib.h>\r
  39 \r
  40 #include <tbb/concurrent_queue.h>\r
  41 \r
  42 #include <boost/property_tree/ptree.hpp>\r
  43 #include <boost/range/algorithm_ext/push_back.hpp>\r
  44 \r
  45 #include <queue>\r
  46 \r
  47 namespace caspar { namespace reroute {\r
  48                 \r
  49 class reroute_producer : public reactive::observer<safe_ptr<const core::data_frame>>\r
  50                                            , public core::frame_producer\r
  51                                            , public enable_safe_from_this<reroute_producer>\r
  52 {\r
  53         const safe_ptr<diagnostics::graph>                                                                      graph_;\r
  54         const safe_ptr<core::frame_factory>                                                                     frame_factory_;\r
  55         \r
  56         tbb::concurrent_bounded_queue<std::shared_ptr<const core::data_frame>>  input_buffer_;\r
  57         std::queue<safe_ptr<core::draw_frame>>                                                          frame_buffer_;\r
  58         safe_ptr<core::draw_frame>                                                                                      last_frame_;\r
  59         uint64_t                                                                                                                        frame_number_;\r
  60 \r
  61 public:\r
  62         explicit reroute_producer(const safe_ptr<core::frame_factory>& frame_factory) \r
  63                 : frame_factory_(frame_factory)\r
  64                 , last_frame_(core::draw_frame::empty())\r
  65                 , frame_number_(0)\r
  66         {\r
  67                 graph_->set_color("late-frame", diagnostics::color(0.6f, 0.3f, 0.3f));\r
  68                 graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));\r
  69                 graph_->set_text(print());\r
  70                 diagnostics::register_graph(graph_);\r
  71 \r
  72                 input_buffer_.set_capacity(1);\r
  73         }\r
  74         \r
  75         // observable\r
  76 \r
  77         void on_next(const safe_ptr<const core::data_frame>& frame)\r
  78         {\r
  79                 if(!input_buffer_.try_push(frame))\r
  80                         graph_->set_tag("dropped-frame");\r
  81         }\r
  82 \r
  83         // frame_producer\r
  84                         \r
  85         virtual safe_ptr<core::draw_frame> receive(int) override\r
  86         {\r
  87                 if(!frame_buffer_.empty())\r
  88                 {\r
  89                         auto frame = frame_buffer_.front();\r
  90                         frame_buffer_.pop();\r
  91                         return last_frame_ = frame;\r
  92                 }\r
  93                 \r
  94                 std::shared_ptr<const core::data_frame> read_frame;\r
  95                 if(input_buffer_.try_pop(read_frame) || read_frame->image_data().empty())\r
  96                 {\r
  97                         graph_->set_tag("late-frame");\r
  98                         return core::draw_frame::late();                \r
  99                 }\r
 100                 \r
 101                 frame_number_++;\r
 102                 \r
 103                 bool double_speed       = std::abs(frame_factory_->get_video_format_desc().fps / 2.0 - read_frame->get_frame_rate()) < 0.01;            \r
 104                 bool half_speed         = std::abs(read_frame->get_frame_rate() / 2.0 - frame_factory_->get_video_format_desc().fps) < 0.01;\r
 105 \r
 106                 if(half_speed && frame_number_ % 2 == 0) // Skip frame\r
 107                         return receive(0);\r
 108 \r
 109                 auto frame = frame_factory_->create_frame(this, read_frame->get_pixel_format_desc());\r
 110 \r
 111                 A_memcpy(frame->image_data(0).begin(), read_frame->image_data().begin(), read_frame->image_data().size());\r
 112                 boost::push_back(frame->audio_data(), read_frame->audio_data());\r
 113 \r
 114                 frame_buffer_.push(frame);      \r
 115                 \r
 116                 if(double_speed)        \r
 117                         frame_buffer_.push(frame);\r
 118 \r
 119                 return receive(0);\r
 120         }       \r
 121 \r
 122         virtual safe_ptr<core::draw_frame> last_frame() const override\r
 123         {\r
 124                 return last_frame_; \r
 125         }       \r
 126 \r
 127         virtual std::wstring print() const override\r
 128         {\r
 129                 return L"reroute[]";\r
 130         }\r
 131 \r
 132         virtual boost::property_tree::wptree info() const override\r
 133         {\r
 134                 boost::property_tree::wptree info;\r
 135                 info.add(L"type", L"rerotue-producer");\r
 136                 return info;\r
 137         }\r
 138 };\r
 139 \r
 140 safe_ptr<core::frame_producer> create_producer(const safe_ptr<core::frame_factory>& frame_factory, reactive::observable<safe_ptr<const core::data_frame>>& o)\r
 141 {\r
 142         auto producer = make_safe<reroute_producer>(frame_factory);\r
 143         o.subscribe(producer);\r
 144         return create_producer_print_proxy(producer);\r
 145 }\r
 146 \r
 147 }}