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 (const file_not_found& e)
 112                         {
 113                                 CASPAR_LOG_CURRENT_EXCEPTION_AT_LEVEL(debug);
 114                                 CASPAR_LOG(error) << get_message_and_context(e) << " Turn on log level debug for stacktrace.";
 115                                 pCurrentCommand->SetReplyString(L"404 " + pCurrentCommand->print() + L" FAILED\r\n");
 116                         }
 117                         catch (const expected_user_error& e)
 118                         {
 119                                 CASPAR_LOG_CURRENT_EXCEPTION_AT_LEVEL(debug);
 120                                 CASPAR_LOG(info) << get_message_and_context(e) << " Check syntax. Turn on log level debug for stacktrace.";
 121                                 pCurrentCommand->SetReplyString(L"403 " + pCurrentCommand->print() + L" FAILED\r\n");
 122                         }
 123                         catch (const user_error& e)
 124                         {
 125                                 CASPAR_LOG_CURRENT_EXCEPTION_AT_LEVEL(debug);
 126                                 CASPAR_LOG(error) << get_message_and_context(e) << " Check syntax. Turn on log level debug for stacktrace.";
 127                                 pCurrentCommand->SetReplyString(L"403 " + pCurrentCommand->print() + L" FAILED\r\n");
 128                         }
 129                         catch (std::out_of_range&)
 130                         {
 131                                 CASPAR_LOG_CURRENT_EXCEPTION_AT_LEVEL(debug);
 132                                 CASPAR_LOG(error) << L"Missing parameter. Check syntax. Turn on log level debug for stacktrace.";
 133                                 pCurrentCommand->SetReplyString(L"402 " + pCurrentCommand->print() + L" FAILED\r\n");
 134                         }
 135                         catch (boost::bad_lexical_cast&)
 136                         {
 137                                 CASPAR_LOG_CURRENT_EXCEPTION_AT_LEVEL(debug);
 138                                 CASPAR_LOG(error) << L"Invalid parameter. Check syntax. Turn on log level debug for stacktrace.";
 139                                 pCurrentCommand->SetReplyString(L"403 " + pCurrentCommand->print() + L" FAILED\r\n");
 140                         }
 141                         catch (...)
 142                         {
 143                                 CASPAR_LOG_CURRENT_EXCEPTION();
 144                                 CASPAR_LOG(error) << "Failed to execute command:" << pCurrentCommand->print();
 145                                 pCurrentCommand->SetReplyString(L"501 " + pCurrentCommand->print() + L" FAILED\r\n");
 146                         }
 147
 148                         pCurrentCommand->SendReply();
 149
 150                         CASPAR_LOG(trace) << "Ready for a new command";
 151
 152                         tbb::spin_mutex::scoped_lock lock(running_command_mutex_);
 153                         running_command_ = false;
 154                 }
 155                 catch(...)
 156                 {
 157                         CASPAR_LOG_CURRENT_EXCEPTION();
 158                 }
 159         });
 160 }
 161
 162 boost::property_tree::wptree AMCPCommandQueue::info() const
 163 {
 164         boost::property_tree::wptree info;
 165
 166         auto name = executor_.name();
 167         info.add(L"name", name);
 168         auto size = executor_.size();
 169         info.add(L"queued", std::max(0u, size));
 170
 171         bool running_command;
 172         std::wstring running_command_name;
 173         std::wstring running_command_params;
 174         int64_t running_command_elapsed;
 175
 176         lock(running_command_mutex_, [&]
 177         {
 178                 running_command = running_command_;
 179
 180                 if (running_command)
 181                 {
 182                         running_command_name = running_command_name_;
 183                         running_command_params = running_command_params_;
 184                         running_command_elapsed = static_cast<int64_t>(
 185                                 running_command_since_.elapsed() * 1000.0);
 186                 }
 187         });
 188
 189         if (running_command)
 190         {
 191                 info.add(L"running.command", running_command_name);
 192                 info.add(L"running.params", running_command_params);
 193                 info.add(L"running.elapsed", running_command_elapsed);
 194         }
 195
 196         return info;
 197 }
 198
 199 boost::property_tree::wptree AMCPCommandQueue::info_all_queues()
 200 {
 201         boost::property_tree::wptree info;
 202         tbb::spin_mutex::scoped_lock lock(get_global_mutex());
 203
 204         for (auto& queue : get_instances())
 205         {
 206                 info.add_child(L"queues.queue", queue.second->info());
 207         }
 208
 209         return info;
 210 }
 211
 212 }}}