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 = []
  67                 {
  68                         tbb::atomic<int> c;
  69                         c = 0;
  70                         return c;
  71                 }();
  72
  73                 ++counter;
  74                 CASPAR_VERIFY(counter < 8);
  75
  76                 auto consumer = new std::shared_ptr<frame_consumer>(std::move(consumer_));
  77                 boost::thread([=]
  78                 {
  79                         std::unique_ptr<std::shared_ptr<frame_consumer>> pointer_guard(consumer);
  80
  81                         auto str = (*consumer)->print();
  82                         try
  83                         {
  84                                 if(!consumer->unique())
  85                                         CASPAR_LOG(trace) << str << L" Not destroyed on asynchronous destruction thread: " << consumer->use_count();
  86                                 else
  87                                         CASPAR_LOG(trace) << str << L" Destroying on asynchronous destruction thread.";
  88                         }
  89                         catch(...){}
  90
  91                         pointer_guard.reset();
  92
  93                 }).detach();
  94         }
  95
  96         std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
  97         virtual void initialize(const struct video_format_desc& format_desc, int channel_index) override        {return consumer_->initialize(format_desc, channel_index);}
  98         std::wstring print() const override                                                                                                                                     {return consumer_->print();}
  99         std::wstring name() const override                                                                                                                                      {return consumer_->name();}
 100         boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
 101         bool has_synchronization_clock() const override                                                                                                         {return consumer_->has_synchronization_clock();}
 102         int buffer_depth() const override                                                                                                                                       {return consumer_->buffer_depth();}
 103         int index() const override                                                                                                                                                      {return consumer_->index();}
 104         monitor::subject& monitor_output() override                                                                                                                     {return consumer_->monitor_output();}
 105 };
 106
 107 class print_consumer_proxy : public frame_consumer
 108 {
 109         std::shared_ptr<frame_consumer> consumer_;
 110 public:
 111         print_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 112                 : consumer_(std::move(consumer))
 113         {
 114                 CASPAR_LOG(info) << consumer_->print() << L" Initialized.";
 115         }
 116
 117         ~print_consumer_proxy()
 118         {
 119                 auto str = consumer_->print();
 120                 CASPAR_LOG(trace) << str << L" Uninitializing.";
 121                 consumer_.reset();
 122                 CASPAR_LOG(info) << str << L" Uninitialized.";
 123         }
 124
 125         std::future<bool> send(const_frame frame) override                                                                                                      {return consumer_->send(std::move(frame));}
 126         virtual void initialize(const struct video_format_desc& format_desc, int channel_index) override        {return consumer_->initialize(format_desc, channel_index);}
 127         std::wstring print() const override                                                                                                                                     {return consumer_->print();}
 128         std::wstring name() const override                                                                                                                                      {return consumer_->name();}
 129         boost::property_tree::wptree info() const override                                                                                                      {return consumer_->info();}
 130         bool has_synchronization_clock() const override                                                                                                         {return consumer_->has_synchronization_clock();}
 131         int buffer_depth() const override                                                                                                                                       {return consumer_->buffer_depth();}
 132         int index() const override                                                                                                                                                      {return consumer_->index();}
 133         monitor::subject& monitor_output() override                                                                                                                     {return consumer_->monitor_output();}
 134 };
 135
 136 class recover_consumer_proxy : public frame_consumer
 137 {
 138         std::shared_ptr<frame_consumer> consumer_;
 139         int                                                             channel_index_  = -1;
 140         video_format_desc                               format_desc_;
 141 public:
 142         recover_consumer_proxy(spl::shared_ptr<frame_consumer>&& consumer)
 143                 : consumer_(std::move(consumer))
 144         {
 145         }
 146
 147         virtual std::future<bool> send(const_frame frame)
 148         {
 149                 try
 150                 {
 151                         return consumer_->send(frame);
 152                 }
 153                 catch(...)
 154                 {
 155                         CASPAR_LOG_CURRENT_EXCEPTION();
 156                         try
 157                         {
 158                                 consumer_->initialize(format_desc_, channel_index_);
 159                                 return consumer_->send(frame);
 160                         }
 161                         catch(...)
 162                         {
 163                                 CASPAR_LOG_CURRENT_EXCEPTION();
 164                                 CASPAR_LOG(error) << print() << " Failed to recover consumer.";
 165                                 return make_ready_future(false);
 166                         }
 167                 }
 168         }
 169
 170         virtual void initialize(const struct video_format_desc& format_desc, int channel_index)
 171         {
 172                 format_desc_    = format_desc;
 173                 channel_index_  = channel_index;
 174                 return consumer_->initialize(format_desc, channel_index);
 175         }
 176
 177         std::wstring print() const override                                                                             {return consumer_->print();}
 178         std::wstring name() const override                                                                              {return consumer_->name();}
 179         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 180         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 181         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 182         int index() const override                                                                                              {return consumer_->index();}
 183         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 184 };
 185
 186 // This class is used to guarantee that audio cadence is correct. This is important for NTSC audio.
 187 class cadence_guard : public frame_consumer
 188 {
 189         spl::shared_ptr<frame_consumer>         consumer_;
 190         std::vector<int>                                        audio_cadence_;
 191         video_format_desc                                       format_desc_;
 192         boost::circular_buffer<std::size_t>     sync_buffer_;
 193 public:
 194         cadence_guard(const spl::shared_ptr<frame_consumer>& consumer)
 195                 : consumer_(consumer)
 196         {
 197         }
 198
 199         void initialize(const video_format_desc& format_desc, int channel_index) override
 200         {
 201                 audio_cadence_  = format_desc.audio_cadence;
 202                 sync_buffer_    = boost::circular_buffer<std::size_t>(format_desc.audio_cadence.size());
 203                 format_desc_    = format_desc;
 204                 consumer_->initialize(format_desc, channel_index);
 205         }
 206
 207         std::future<bool> send(const_frame frame) override
 208         {
 209                 if(audio_cadence_.size() == 1)
 210                         return consumer_->send(frame);
 211
 212                 std::future<bool> result = make_ready_future(true);
 213
 214                 if(boost::range::equal(sync_buffer_, audio_cadence_) && audio_cadence_.front() * format_desc_.audio_channels == static_cast<int>(frame.audio_data().size()))
 215                 {
 216                         // Audio sent so far is in sync, now we can send the next chunk.
 217                         result = consumer_->send(frame);
 218                         boost::range::rotate(audio_cadence_, std::begin(audio_cadence_)+1);
 219                 }
 220                 else
 221                         CASPAR_LOG(trace) << print() << L" Syncing audio.";
 222
 223                 sync_buffer_.push_back(static_cast<int>(frame.audio_data().size() / format_desc_.audio_channels));
 224
 225                 return std::move(result);
 226         }
 227
 228         std::wstring print() const override                                                                             {return consumer_->print();}
 229         std::wstring name() const override                                                                              {return consumer_->name();}
 230         boost::property_tree::wptree info() const override                                              {return consumer_->info();}
 231         bool has_synchronization_clock() const override                                                 {return consumer_->has_synchronization_clock();}
 232         int buffer_depth() const override                                                                               {return consumer_->buffer_depth();}
 233         int index() const override                                                                                              {return consumer_->index();}
 234         monitor::subject& monitor_output() override                                                             {return consumer_->monitor_output();}
 235 };
 236
 237 spl::shared_ptr<core::frame_consumer> create_consumer(
 238                 const std::vector<std::wstring>& params, interaction_sink* sink)
 239 {
 240         if(params.empty())
 241                 CASPAR_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
 242
 243         auto consumer = frame_consumer::empty();
 244         std::any_of(g_consumer_factories.begin(), g_consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
 245                 {
 246                         try
 247                         {
 248                                 consumer = factory(params, sink);
 249                         }
 250                         catch(...)
 251                         {
 252                                 CASPAR_LOG_CURRENT_EXCEPTION();
 253                         }
 254                         return consumer != frame_consumer::empty();
 255                 });
 256
 257         if(consumer == frame_consumer::empty())
 258                 CASPAR_THROW_EXCEPTION(file_not_found() << msg_info("No match found for supplied commands. Check syntax."));
 259
 260         return spl::make_shared<destroy_consumer_proxy>(
 261                         spl::make_shared<print_consumer_proxy>(
 262                          spl::make_shared<recover_consumer_proxy>(
 263                           spl::make_shared<cadence_guard>(
 264                            std::move(consumer)))));
 265 }
 266
 267 spl::shared_ptr<frame_consumer> create_consumer(
 268                 const std::wstring& element_name,
 269                 const boost::property_tree::wptree& element,
 270                 interaction_sink* sink)
 271 {
 272         auto found = g_preconfigured_consumer_factories.find(element_name);
 273
 274         if (found == g_preconfigured_consumer_factories.end())
 275                 CASPAR_THROW_EXCEPTION(file_not_found()
 276                         << msg_info(L"No consumer factory registered for element name " + element_name));
 277
 278         return spl::make_shared<destroy_consumer_proxy>(
 279                         spl::make_shared<print_consumer_proxy>(
 280                                         spl::make_shared<recover_consumer_proxy>(
 281                                                         spl::make_shared<cadence_guard>(
 282                                                                         found->second(element, sink)))));
 283 }
 284
 285 const spl::shared_ptr<frame_consumer>& frame_consumer::empty()
 286 {
 287         class empty_frame_consumer : public frame_consumer
 288         {
 289         public:
 290                 std::future<bool> send(const_frame) override { return make_ready_future(false); }
 291                 void initialize(const video_format_desc&, int) override{}
 292                 std::wstring print() const override {return L"empty";}
 293                 std::wstring name() const override {return L"empty";}
 294                 bool has_synchronization_clock() const override {return false;}
 295                 int buffer_depth() const override {return 0;};
 296                 virtual int index() const{return -1;}
 297                 monitor::subject& monitor_output() override {static monitor::subject monitor_subject(""); return monitor_subject;}
 298                 boost::property_tree::wptree info() const override
 299                 {
 300                         boost::property_tree::wptree info;
 301                         info.add(L"type", L"empty");
 302                         return info;
 303                 }
 304         };
 305         static spl::shared_ptr<frame_consumer> consumer = spl::make_shared<empty_frame_consumer>();
 306         return consumer;
 307 }
 308
 309 }}