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