git.sesse.net Git - casparcg/blob - protocol/amcp/AMCPCommandQueue.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: Nicklas P Andersson
  20 */
  21
  22 #include "../StdAfx.h"
  23
  24 #include "AMCPCommandQueue.h"
  25
  26 #include <common/lock.h>
  27
  28 #include <boost/property_tree/ptree.hpp>
  29
  30 #include <cmath>
  31
  32 namespace caspar { namespace protocol { namespace amcp {
  33
  34 namespace {
  35
  36 tbb::spin_mutex& get_global_mutex()
  37 {
  38         static tbb::spin_mutex mutex;
  39
  40         return mutex;
  41 }
  42
  43 std::map<std::wstring, AMCPCommandQueue*>& get_instances()
  44 {
  45         static std::map<std::wstring, AMCPCommandQueue*> queues;
  46
  47         return queues;
  48 }
  49
  50 }
  51
  52 AMCPCommandQueue::AMCPCommandQueue(const std::wstring& name)
  53         : executor_(L"AMCPCommandQueue " + name)
  54 {
  55         tbb::spin_mutex::scoped_lock lock(get_global_mutex());
  56
  57         get_instances().insert(std::make_pair(name, this));
  58 }
  59
  60 AMCPCommandQueue::~AMCPCommandQueue()
  61 {
  62         tbb::spin_mutex::scoped_lock lock(get_global_mutex());
  63
  64         get_instances().erase(executor_.name());
  65 }
  66
  67 void AMCPCommandQueue::AddCommand(AMCPCommand::ptr_type pCurrentCommand)
  68 {
  69         if(!pCurrentCommand)
  70                 return;
  71
  72         if(executor_.size() > 128)
  73         {
  74                 try
  75                 {
  76                         CASPAR_LOG(error) << "AMCP Command Queue Overflow.";
  77                         CASPAR_LOG(error) << "Failed to execute command:" << pCurrentCommand->print();
  78                         pCurrentCommand->SetReplyString(L"500 FAILED\r\n");
  79                         pCurrentCommand->SendReply();
  80                 }
  81                 catch(...)
  82                 {
  83                         CASPAR_LOG_CURRENT_EXCEPTION();
  84                 }
  85         }
  86
  87         executor_.begin_invoke([=]
  88         {
  89                 try
  90                 {
  91                         try
  92                         {
  93                                 caspar::timer timer;
  94
  95                                 auto print = pCurrentCommand->print();
  96                                 auto params = boost::join(pCurrentCommand->parameters(), L" ");
  97
  98                                 {
  99                                         tbb::spin_mutex::scoped_lock lock(running_command_mutex_);
 100                                         running_command_ = true;
 101                                         running_command_name_ = print;
 102                                         running_command_params_ = std::move(params);
 103                                         running_command_since_.restart();
 104                                 }
 105
 106                                 if (pCurrentCommand->Execute())
 107                                         CASPAR_LOG(debug) << "Executed command (" << timer.elapsed() << "s): " << print;
 108                                 else
 109                                         CASPAR_LOG(warning) << "Failed to execute command: " << print;
 110                         }
 111                         catch (file_not_found&)
 112                         {
 113                                 CASPAR_LOG(error) << L"File not found. No match found for parameters. Check syntax.";
 114                                 pCurrentCommand->SetReplyString(L"404 " + pCurrentCommand->print() + L" FAILED\r\n");
 115                         }
 116                         catch (const user_error& e)
 117                         {
 118                                 CASPAR_LOG(error) << *boost::get_error_info<msg_info_t>(e) << ". Check syntax.";
 119                                 pCurrentCommand->SetReplyString(L"403 " + pCurrentCommand->print() + L" FAILED\r\n");
 120                         }
 121                         catch (std::out_of_range&)
 122                         {
 123                                 CASPAR_LOG(error) << L"Missing parameter. Check syntax.";
 124                                 pCurrentCommand->SetReplyString(L"402 " + pCurrentCommand->print() + L" FAILED\r\n");
 125                         }
 126                         catch (...)
 127                         {
 128                                 CASPAR_LOG_CURRENT_EXCEPTION();
 129                                 CASPAR_LOG(warning) << "Failed to execute command:" << pCurrentCommand->print();
 130                                 pCurrentCommand->SetReplyString(L"501 " + pCurrentCommand->print() + L" FAILED\r\n");
 131                         }
 132
 133                         pCurrentCommand->SendReply();
 134
 135                         CASPAR_LOG(trace) << "Ready for a new command";
 136
 137                         tbb::spin_mutex::scoped_lock lock(running_command_mutex_);
 138                         running_command_ = false;
 139                 }
 140                 catch(...)
 141                 {
 142                         CASPAR_LOG_CURRENT_EXCEPTION();
 143                 }
 144         });
 145 }
 146
 147 boost::property_tree::wptree AMCPCommandQueue::info() const
 148 {
 149         boost::property_tree::wptree info;
 150
 151         auto name = executor_.name();
 152         info.add(L"name", name);
 153         auto size = executor_.size();
 154         info.add(L"queued", std::max(0u, size));
 155
 156         bool running_command;
 157         std::wstring running_command_name;
 158         std::wstring running_command_params;
 159         int64_t running_command_elapsed;
 160
 161         lock(running_command_mutex_, [&]
 162         {
 163                 running_command = running_command_;
 164
 165                 if (running_command)
 166                 {
 167                         running_command_name = running_command_name_;
 168                         running_command_params = running_command_params_;
 169                         running_command_elapsed = static_cast<int64_t>(
 170                                 running_command_since_.elapsed() * 1000.0);
 171                 }
 172         });
 173
 174         if (running_command)
 175         {
 176                 info.add(L"running.command", running_command_name);
 177                 info.add(L"running.params", running_command_params);
 178                 info.add(L"running.elapsed", running_command_elapsed);
 179         }
 180
 181         return info;
 182 }
 183
 184 boost::property_tree::wptree AMCPCommandQueue::info_all_queues()
 185 {
 186         boost::property_tree::wptree info;
 187         tbb::spin_mutex::scoped_lock lock(get_global_mutex());
 188
 189         for (auto& queue : get_instances())
 190         {
 191                 info.add_child(L"queues.queue", queue.second->info());
 192         }
 193
 194         return info;
 195 }
 196
 197 }}}