]> git.sesse.net Git - casparcg/blob - modules/reroute/producer/reroute_producer.cpp
2.1.0: Integrating monitoring.
[casparcg] / 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/frame/write_frame.h>\r
31 #include <core/frame/data_frame.h>\r
32 \r
33 #include <common/except.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<spl::shared_ptr<const core::data_frame>>\r
50                                            , public core::frame_producer\r
51 {\r
52         const spl::shared_ptr<diagnostics::graph>                                                                       graph_;\r
53         const spl::shared_ptr<core::frame_factory>                                                                      frame_factory_;\r
54         \r
55         tbb::concurrent_bounded_queue<std::shared_ptr<const core::data_frame>>  input_buffer_;\r
56         std::queue<spl::shared_ptr<core::draw_frame>>                                                           frame_buffer_;\r
57         spl::shared_ptr<core::draw_frame>                                                                                       last_frame_;\r
58         uint64_t                                                                                                                        frame_number_;\r
59 \r
60 public:\r
61         explicit reroute_producer(const spl::shared_ptr<core::frame_factory>& frame_factory) \r
62                 : frame_factory_(frame_factory)\r
63                 , last_frame_(core::draw_frame::empty())\r
64                 , frame_number_(0)\r
65         {\r
66                 graph_->set_color("late-frame", diagnostics::color(0.6f, 0.3f, 0.3f));\r
67                 graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));\r
68                 graph_->set_text(print());\r
69                 diagnostics::register_graph(graph_);\r
70 \r
71                 input_buffer_.set_capacity(1);\r
72         }\r
73         \r
74         // observable\r
75 \r
76         void on_next(const spl::shared_ptr<const core::data_frame>& frame)\r
77         {\r
78                 if(!input_buffer_.try_push(frame))\r
79                         graph_->set_tag("dropped-frame");\r
80         }\r
81 \r
82         // frame_producer\r
83                         \r
84         virtual spl::shared_ptr<core::draw_frame> receive(int) override\r
85         {\r
86                 if(!frame_buffer_.empty())\r
87                 {\r
88                         auto frame = frame_buffer_.front();\r
89                         frame_buffer_.pop();\r
90                         return last_frame_ = frame;\r
91                 }\r
92                 \r
93                 std::shared_ptr<const core::data_frame> read_frame;\r
94                 if(input_buffer_.try_pop(read_frame) || read_frame->image_data().empty())\r
95                 {\r
96                         graph_->set_tag("late-frame");\r
97                         return core::draw_frame::late();                \r
98                 }\r
99                 \r
100                 frame_number_++;\r
101                 \r
102                 bool double_speed       = std::abs(frame_factory_->get_video_format_desc().fps / 2.0 - read_frame->get_frame_rate()) < 0.01;            \r
103                 bool half_speed         = std::abs(read_frame->get_frame_rate() / 2.0 - frame_factory_->get_video_format_desc().fps) < 0.01;\r
104 \r
105                 if(half_speed && frame_number_ % 2 == 0) // Skip frame\r
106                         return receive(0);\r
107 \r
108                 auto frame = frame_factory_->create_frame(this, read_frame->get_pixel_format_desc());\r
109 \r
110                 A_memcpy(frame->image_data(0).begin(), read_frame->image_data().begin(), read_frame->image_data().size());\r
111                 boost::push_back(frame->audio_data(), read_frame->audio_data());\r
112 \r
113                 frame_buffer_.push(frame);      \r
114                 \r
115                 if(double_speed)        \r
116                         frame_buffer_.push(frame);\r
117 \r
118                 return receive(0);\r
119         }       \r
120 \r
121         virtual spl::shared_ptr<core::draw_frame> last_frame() const override\r
122         {\r
123                 return last_frame_; \r
124         }       \r
125 \r
126         virtual std::wstring print() const override\r
127         {\r
128                 return L"reroute[]";\r
129         }\r
130 \r
131         virtual std::wstring name() const override\r
132         {\r
133                 return L"reroute";\r
134         }\r
135 \r
136         virtual boost::property_tree::wptree info() const override\r
137         {\r
138                 boost::property_tree::wptree info;\r
139                 info.add(L"type", L"rerotue-producer");\r
140                 return info;\r
141         }\r
142 };\r
143 \r
144 spl::shared_ptr<core::frame_producer> create_producer(const spl::shared_ptr<core::frame_factory>& frame_factory, reactive::observable<spl::shared_ptr<const core::data_frame>>& o)\r
145 {\r
146         auto producer = spl::make_shared<reroute_producer>(frame_factory);\r
147         o.subscribe(producer);\r
148         return core::wrap_producer(producer);\r
149 }\r
150 \r
151 }}