git.sesse.net Git - casparcg/blob - core/consumer/frame_consumer.cpp

   1 /*
   2 * Copyright (c) 2011 Sveriges Television AB <info@casparcg.com>
   3 *
   4 * This file is part of CasparCG (www.casparcg.com).
   5 *
   6 * CasparCG is free software: you can redistribute it and/or modify
   7 * it under the terms of the GNU General Public License as published by
   8 * the Free Software Foundation, either version 3 of the License, or
   9 * (at your option) any later version.
  10 *
  11 * CasparCG is distributed in the hope that it will be useful,
  12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 * GNU General Public License for more details.
  15 *
  16 * You should have received a copy of the GNU General Public License
  17 * along with CasparCG. If not, see <http://www.gnu.org/licenses/>.
  18 *
  19 * Author: Robert Nagy, ronag89@gmail.com
  20 */
  21
  22 #include "../StdAfx.h"
  23
  24 #include "frame_consumer.h"
  25
  26 #include <common/except.h>
  27 #include <common/future.h>
  28
  29 #include <core/video_format.h>
  30 #include <core/frame/frame.h>
  31
  32 #include <boost/thread.hpp>
  33
  34 #include <future>
  35 #include <vector>
  36 #include <map>
  37
  38 namespace caspar { namespace core {
  39
  40 std::vector<consumer_factory_t> g_consumer_factories;
  41 std::map<std::wstring, preconfigured_consumer_factory_t> g_preconfigured_consumer_factories;
  42
  43 void register_consumer_factory(const consumer_factory_t& factory)
  44 {
  45         g_consumer_factories.push_back(factory);
  46 }
  47
  48 void register_preconfigured_consumer_factory(
  49                 const std::wstring& element_name,
  50                 const preconfigured_consumer_factory_t& factory)
  51 {
  52         g_preconfigured_consumer_factories.insert(std::make_pair(element_name, factory));
  53 }
  54
  55 class destroy_consumer_proxy : public frame_consumer
  56 {
  57         std::shared_ptr<frame_consumer> consumer_;
  58 public:
  59         destroy_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
  60                 : consumer_(std::move(consumer))
  61         {
  62         }
  63
  64         ~destroy_consumer_proxy()
  65         {
  66                 static tbb::atomic<int> counter = tbb::atomic<int>();
  67
  68                 ++counter;
  69                 CASPAR_VERIFY(counter < 8);
  70
  71                 auto consumer = new std::shared_ptr<frame_consumer>(std::move(consumer_));
  72                 boost::thread([=]
  73                 {
  74                         std::unique_ptr<std::shared_ptr<frame_consumer>> pointer_guard(consumer);
  75
  76                         auto str = (*consumer)->print();
  77                         try
  78                         {
  79                                 if(!consumer->unique())
  80                                         CASPAR_LOG(trace) << str << L" Not destroyed on asynchronous destruction thread: " << consumer->use_count();
  81                                 else
  82                                         CASPAR_LOG(trace) << str << L" Destroying on asynchronous destruction thread.";
  83                         }
  84                         catch(...){}
  85
  86                         pointer_guard.reset();
  87
  88                         --counter;
  89                 }).detach();
  90         }
  91
  92         std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
  93         virtual void initialize(const struct video_format_desc& format_desc, int channel_index) override        {return consumer_->initialize(format_desc, channel_index);}
  94         std::wstring print() const override                                                                                                                                     {return consumer_->print();}
  95         std::wstring name() const override                                                                                                                                      {return consumer_->name();}
  96         boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
  97         bool has_synchronization_clock() const override                                                                                                         {return consumer_->has_synchronization_clock();}
  98         int buffer_depth() const override                                                                                                                                       {return consumer_->buffer_depth();}
  99         int index() const override                                                                                                                                                      {return consumer_->index();}
 100         monitor::subject& monitor_output() override                                                                                                                     {return consumer_->monitor_output();}
 101 };
 102
 103 class print_consumer_proxy : public frame_consumer
 104 {
 105         std::shared_ptr<frame_consumer> consumer_;
 106 public:
 107         print_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 108                 : consumer_(std::move(consumer))
 109         {
 110                 CASPAR_LOG(info) << consumer_->print() << L" Initialized.";
 111         }
 112
 113         ~print_consumer_proxy()
 114         {
 115                 auto str = consumer_->print();
 116                 CASPAR_LOG(trace) << str << L" Uninitializing.";
 117                 consumer_.reset();
 118                 CASPAR_LOG(info) << str << L" Uninitialized.";
 119         }
 120
 121         std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
 122         virtual void initialize(const struct video_format_desc& format_desc, int channel_index) override        {return consumer_->initialize(format_desc, channel_index);}
 123         std::wstring print() const override                                                                                                                                     {return consumer_->print();}
 124         std::wstring name() const override                                                                                                                                      {return consumer_->name();}
 125         boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
 126         bool has_synchronization_clock() const override                                                                                                         {return consumer_->has_synchronization_clock();}
 127         int buffer_depth() const override                                                                                                                                       {return consumer_->buffer_depth();}
 128         int index() const override                                                                                                                                                      {return consumer_->index();}
 129         monitor::subject& monitor_output() override                                                                                                                     {return consumer_->monitor_output();}
 130 };
 131
 132 class recover_consumer_proxy : public frame_consumer
 133 {
 134         std::shared_ptr<frame_consumer> consumer_;
 135         int                                                             channel_index_  = -1;
 136         video_format_desc                               format_desc_;
 137 public:
 138         recover_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 139                 : consumer_(std::move(consumer))
 140         {
 141         }
 142
 143         virtual std::future<bool> send(const_frame frame)
 144         {
 145                 try
 146                 {
 147                         return consumer_->send(frame);
 148                 }
 149                 catch(...)
 150                 {
 151                         CASPAR_LOG_CURRENT_EXCEPTION();
 152                         try
 153                         {
 154                                 consumer_->initialize(format_desc_, channel_index_);
 155                                 return consumer_->send(frame);
 156                         }
 157                         catch(...)
 158                         {
 159                                 CASPAR_LOG_CURRENT_EXCEPTION();
 160                                 CASPAR_LOG(error) << print() << " Failed to recover consumer.";
 161                                 return make_ready_future(false);
 162                         }
 163                 }
 164         }
 165
 166         virtual void initialize(const struct video_format_desc& format_desc, int channel_index)
 167         {
 168                 format_desc_    = format_desc;
 169                 channel_index_  = channel_index;
 170                 return consumer_->initialize(format_desc, channel_index);
 171         }
 172
 173         std::wstring print() const override                                                                             {return consumer_->print();}
 174         std::wstring name() const override                                                                              {return consumer_->name();}
 175         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 176         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 177         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 178         int index() const override                                                                                              {return consumer_->index();}
 179         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 180 };
 181
 182 // This class is used to guarantee that audio cadence is correct. This is important for NTSC audio.
 183 class cadence_guard : public frame_consumer
 184 {
 185         spl::shared_ptr<frame_consumer>         consumer_;
 186         std::vector<int>                                        audio_cadence_;
 187         video_format_desc                                       format_desc_;
 188         boost::circular_buffer<std::size_t>     sync_buffer_;
 189 public:
 190         cadence_guard(const spl::shared_ptr<frame_consumer>& consumer)
 191                 : consumer_(consumer)
 192         {
 193         }
 194
 195         void initialize(const video_format_desc& format_desc, int channel_index) override
 196         {
 197                 audio_cadence_  = format_desc.audio_cadence;
 198                 sync_buffer_    = boost::circular_buffer<std::size_t>(format_desc.audio_cadence.size());
 199                 format_desc_    = format_desc;
 200                 consumer_->initialize(format_desc, channel_index);
 201         }
 202
 203         std::future<bool> send(const_frame frame) override
 204         {
 205                 if(audio_cadence_.size() == 1)
 206                         return consumer_->send(frame);
 207
 208                 std::future<bool> result = make_ready_future(true);
 209
 210                 if(boost::range::equal(sync_buffer_, audio_cadence_) && audio_cadence_.front() * format_desc_.audio_channels == static_cast<int>(frame.audio_data().size()))
 211                 {
 212                         // Audio sent so far is in sync, now we can send the next chunk.
 213                         result = consumer_->send(frame);
 214                         boost::range::rotate(audio_cadence_, std::begin(audio_cadence_)+1);
 215                 }
 216                 else
 217                         CASPAR_LOG(trace) << print() << L" Syncing audio.";
 218
 219                 sync_buffer_.push_back(static_cast<int>(frame.audio_data().size() / format_desc_.audio_channels));
 220
 221                 return std::move(result);
 222         }
 223
 224         std::wstring print() const override                                                                             {return consumer_->print();}
 225         std::wstring name() const override                                                                              {return consumer_->name();}
 226         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 227         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 228         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 229         int index() const override                                                                                              {return consumer_->index();}
 230         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 231 };
 232
 233 spl::shared_ptr<core::frame_consumer> create_consumer(
 234                 const std::vector<std::wstring>& params, interaction_sink* sink)
 235 {
 236         if(params.empty())
 237                 CASPAR_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
 238
 239         auto consumer = frame_consumer::empty();
 240         std::any_of(g_consumer_factories.begin(), g_consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
 241                 {
 242                         try
 243                         {
 244                                 consumer = factory(params, sink);
 245                         }
 246                         catch(...)
 247                         {
 248                                 CASPAR_LOG_CURRENT_EXCEPTION();
 249                         }
 250                         return consumer != frame_consumer::empty();
 251                 });
 252
 253         if(consumer == frame_consumer::empty())
 254                 CASPAR_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax."));
 255
 256         return spl::make_shared<destroy_consumer_proxy>(
 257                         spl::make_shared<print_consumer_proxy>(
 258                          spl::make_shared<recover_consumer_proxy>(
 259                           spl::make_shared<cadence_guard>(
 260                            std::move(consumer)))));
 261 }
 262
 263 spl::shared_ptr<frame_consumer> create_consumer(
 264                 const std::wstring& element_name,
 265                 const boost::property_tree::wptree& element,
 266                 interaction_sink* sink)
 267 {
 268         auto found = g_preconfigured_consumer_factories.find(element_name);
 269
 270         if (found == g_preconfigured_consumer_factories.end())
 271                 CASPAR_THROW_EXCEPTION(file_not_found()
 272                         << msg_info(L"No consumer factory registered for element name " + element_name));
 273
 274         return spl::make_shared<destroy_consumer_proxy>(
 275                         spl::make_shared<print_consumer_proxy>(
 276                                         spl::make_shared<recover_consumer_proxy>(
 277                                                         spl::make_shared<cadence_guard>(
 278                                                                         found->second(element, sink)))));
 279 }
 280
 281 const spl::shared_ptr<frame_consumer>& frame_consumer::empty()
 282 {
 283         class empty_frame_consumer : public frame_consumer
 284         {
 285         public:
 286                 std::future<bool> send(const_frame) override { return make_ready_future(false); }
 287                 void initialize(const video_format_desc&, int) override{}
 288                 std::wstring print() const override {return L"empty";}
 289                 std::wstring name() const override {return L"empty";}
 290                 bool has_synchronization_clock() const override {return false;}
 291                 int buffer_depth() const override {return 0;};
 292                 virtual int index() const{return -1;}
 293                 monitor::subject& monitor_output() override {static monitor::subject monitor_subject(""); return monitor_subject;}
 294                 boost::property_tree::wptree info() const override
 295                 {
 296                         boost::property_tree::wptree info;
 297                         info.add(L"type", L"empty");
 298                         return info;
 299                 }
 300         };
 301         static spl::shared_ptr<frame_consumer> consumer = spl::make_shared<empty_frame_consumer>();
 302         return consumer;
 303 }
 304
 305 }}