git.sesse.net Git - casparcg/blob - protocol/util/AsyncEventServer.cpp

   1 /*\r
   2 * copyright (c) 2010 Sveriges Television AB <info@casparcg.com>\r
   3 *\r
   4 *  This file is part of CasparCG.\r
   5 *\r
   6 *    CasparCG is free software: you can redistribute it and/or modify\r
   7 *    it under the terms of the GNU General Public License as published by\r
   8 *    the Free Software Foundation, either version 3 of the License, or\r
   9 *    (at your option) any later version.\r
  10 *\r
  11 *    CasparCG is distributed in the hope that it will be useful,\r
  12 *    but WITHOUT ANY WARRANTY; without even the implied warranty of\r
  13 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the\r
  14 *    GNU General Public License for more details.\r
  15 \r
  16 *    You should have received a copy of the GNU General Public License\r
  17 *    along with CasparCG.  If not, see <http://www.gnu.org/licenses/>.\r
  18 *\r
  19 */\r
  20  \r
  21 // AsyncEventServer.cpp: implementation of the AsyncEventServer class.\r
  22 //\r
  23 //////////////////////////////////////////////////////////////////////\r
  24 \r
  25 #include "..\stdafx.h"\r
  26 \r
  27 #include "AsyncEventServer.h"\r
  28 \r
  29 #include "ProtocolStrategy.h"\r
  30 \r
  31 #include <algorithm>\r
  32 #include <array>\r
  33 #include <string>\r
  34 #include <set>\r
  35 #include <vector>\r
  36 \r
  37 #include <boost/asio.hpp>\r
  38 #include <boost/thread.hpp>\r
  39 #include <boost/lexical_cast.hpp>\r
  40 \r
  41 using boost::asio::ip::tcp;\r
  42 \r
  43 namespace caspar { namespace IO {\r
  44         \r
  45 bool ConvertWideCharToMultiByte(UINT codePage, const std::wstring& wideString, std::vector<char>& destBuffer)\r
  46 {\r
  47         int bytesWritten = 0;\r
  48         int multibyteBufferCapacity = WideCharToMultiByte(codePage, 0, wideString.c_str(), static_cast<int>(wideString.length()), 0, 0, NULL, NULL);\r
  49         if(multibyteBufferCapacity > 0) \r
  50         {\r
  51                 destBuffer.resize(multibyteBufferCapacity);\r
  52                 bytesWritten = WideCharToMultiByte(codePage, 0, wideString.c_str(), static_cast<int>(wideString.length()), destBuffer.data(), static_cast<int>(destBuffer.size()), NULL, NULL);\r
  53         }\r
  54         destBuffer.resize(bytesWritten);\r
  55         return (bytesWritten > 0);\r
  56 }\r
  57 \r
  58 bool ConvertMultiByteToWideChar(UINT codePage, char* pSource, int sourceLength, std::vector<wchar_t>& wideBuffer, int& countLeftovers)\r
  59 {\r
  60         if(codePage == CP_UTF8) {\r
  61                 countLeftovers = 0;\r
  62                 //check from the end of pSource for ev. uncompleted UTF-8 byte sequence\r
  63                 if(pSource[sourceLength-1] & 0x80) {\r
  64                         //The last byte is part of a multibyte sequence. If the sequence is not complete, we need to save the partial sequence\r
  65                         int bytesToCheck = std::min<int>(4, sourceLength);      //a sequence contains a maximum of 4 bytes\r
  66                         int currentLeftoverIndex = sourceLength-1;\r
  67                         for(; bytesToCheck > 0; --bytesToCheck, --currentLeftoverIndex) {\r
  68                                 ++countLeftovers;\r
  69                                 if(pSource[currentLeftoverIndex] & 0x80) {\r
  70                                         if(pSource[currentLeftoverIndex] & 0x40) { //The two high-bits are set, this is the "header"\r
  71                                                 int expectedSequenceLength = 2;\r
  72                                                 if(pSource[currentLeftoverIndex] & 0x20)\r
  73                                                         ++expectedSequenceLength;\r
  74                                                 if(pSource[currentLeftoverIndex] & 0x10)\r
  75                                                         ++expectedSequenceLength;\r
  76 \r
  77                                                 if(countLeftovers < expectedSequenceLength) {\r
  78                                                         //The sequence is incomplete. Leave the leftovers to be interpreted with the next call\r
  79                                                         break;\r
  80                                                 }\r
  81                                                 //The sequence is complete, there are no leftovers. \r
  82                                                 //...OR...\r
  83                                                 //error. Let the conversion-function take the hit.\r
  84                                                 countLeftovers = 0;\r
  85                                                 break;\r
  86                                         }\r
  87                                 }\r
  88                                 else {\r
  89                                         //error. Let the conversion-function take the hit.\r
  90                                         countLeftovers = 0;\r
  91                                         break;\r
  92                                 }\r
  93                         }\r
  94                         if(countLeftovers == 4) {\r
  95                                 //error. Let the conversion-function take the hit.\r
  96                                 countLeftovers = 0;\r
  97                         }\r
  98                 }\r
  99         }\r
 100 \r
 101         int charsWritten = 0;\r
 102         int sourceBytesToProcess = sourceLength-countLeftovers;\r
 103         int wideBufferCapacity = MultiByteToWideChar(codePage, 0, pSource, sourceBytesToProcess, NULL, NULL);\r
 104         if(wideBufferCapacity > 0) \r
 105         {\r
 106                 wideBuffer.resize(wideBufferCapacity);\r
 107                 charsWritten = MultiByteToWideChar(codePage, 0, pSource, sourceBytesToProcess, wideBuffer.data(), static_cast<int>(wideBuffer.size()));\r
 108         }\r
 109         //copy the leftovers to the front of the buffer\r
 110         if(countLeftovers > 0) {\r
 111                 memcpy(pSource, &(pSource[sourceBytesToProcess]), countLeftovers);\r
 112         }\r
 113 \r
 114         wideBuffer.resize(charsWritten);\r
 115         return (charsWritten > 0);\r
 116 }\r
 117 \r
 118 class connection;\r
 119 \r
 120 typedef std::set<spl::shared_ptr<connection>> connection_set;\r
 121 \r
 122 class connection : public spl::enable_shared_from_this<connection>, public ClientInfo\r
 123 {    \r
 124     spl::shared_ptr<tcp::socket>                socket_; \r
 125         const std::wstring                                      name_;\r
 126 \r
 127         std::array<char, 8192>                          data_;\r
 128 \r
 129         std::vector<char>                                       buffer_;\r
 130 \r
 131         const spl::shared_ptr<IProtocolStrategy>        protocol_;\r
 132         spl::shared_ptr<connection_set>                         connection_set_;\r
 133 \r
 134 public:\r
 135     static spl::shared_ptr<connection> create(spl::shared_ptr<tcp::socket> socket, const ProtocolStrategyPtr& protocol, spl::shared_ptr<connection_set> connection_set)\r
 136         {\r
 137                 spl::shared_ptr<connection> con(new connection(socket, protocol, connection_set));\r
 138                 con->read_some();\r
 139                 return con;\r
 140     }\r
 141 \r
 142         std::wstring print() const\r
 143         {\r
 144                 return L"[" + name_ + L"]";\r
 145         }\r
 146         \r
 147         /* ClientInfo */\r
 148         \r
 149         virtual void Send(const std::wstring& data)\r
 150         {\r
 151                 write_some(data);\r
 152         }\r
 153 \r
 154         virtual void Disconnect()\r
 155         {\r
 156                 stop();\r
 157         }\r
 158         \r
 159         /**************/\r
 160         \r
 161         void stop()\r
 162         {\r
 163                 connection_set_->erase(shared_from_this());\r
 164                 socket_->shutdown(boost::asio::socket_base::shutdown_both);\r
 165                 socket_->close();\r
 166                 CASPAR_LOG(info) << print() << L" Disconnected.";\r
 167         }\r
 168 \r
 169 private:\r
 170     connection(const spl::shared_ptr<tcp::socket>& socket, const ProtocolStrategyPtr& protocol, const spl::shared_ptr<connection_set>& connection_set) \r
 171                 : socket_(socket)\r
 172                 , name_((socket_->is_open() ? u16(socket_->local_endpoint().address().to_string() + ":" + boost::lexical_cast<std::string>(socket_->local_endpoint().port())) : L"no-address"))\r
 173                 , protocol_(protocol)\r
 174                 , connection_set_(connection_set)\r
 175         {\r
 176                 CASPAR_LOG(info) << print() << L" Connected.";\r
 177     }\r
 178                         \r
 179     void handle_read(const boost::system::error_code& error, size_t bytes_transferred) \r
 180         {               \r
 181                 if(!error)\r
 182                 {\r
 183                         try\r
 184                         {\r
 185                                 CASPAR_LOG(trace) << print() << L" Received: " << u16(std::string(data_.begin(), data_.begin() + bytes_transferred));\r
 186 \r
 187                                 buffer_.insert(buffer_.end(), data_.begin(), data_.begin() + bytes_transferred);\r
 188                 \r
 189                                 std::vector<wchar_t> str;\r
 190                                 int left_overs;\r
 191                                 ConvertMultiByteToWideChar(protocol_->GetCodepage(), buffer_.data(), static_cast<int>(buffer_.size()), str, left_overs);\r
 192                                 buffer_.resize(left_overs);\r
 193                 \r
 194                                 protocol_->Parse(str.data(), static_cast<int>(str.size()), shared_from_this());\r
 195                         }\r
 196                         catch(...)\r
 197                         {\r
 198                                 CASPAR_LOG_CURRENT_EXCEPTION();\r
 199                         }\r
 200                         \r
 201                         read_some();\r
 202                 }  \r
 203                 else if (error != boost::asio::error::operation_aborted)\r
 204                         stop();         \r
 205                 else\r
 206                         read_some();\r
 207     }\r
 208 \r
 209     void handle_write(const spl::shared_ptr<std::vector<char>>& data, const boost::system::error_code& error, size_t bytes_transferred)\r
 210         {\r
 211                 if(!error)                      \r
 212                 {\r
 213                         if(data->size() < 512)\r
 214                                 CASPAR_LOG(trace) << print() << L" Sent: " << u16(std::string(data->begin(), data->end()));\r
 215                         else\r
 216                                 CASPAR_LOG(trace) << print() << L" Sent more than 512 bytes.";\r
 217                 }\r
 218                 else if (error != boost::asio::error::operation_aborted)                \r
 219                         stop();         \r
 220     }\r
 221 \r
 222         void read_some()\r
 223         {\r
 224                 socket_->async_read_some(boost::asio::buffer(data_.data(), data_.size()), std::bind(&connection::handle_read, shared_from_this(), std::placeholders::_1, std::placeholders::_2));\r
 225         }\r
 226         \r
 227         void write_some(const std::wstring& data)\r
 228         {\r
 229                 spl::shared_ptr<std::vector<char>> str;\r
 230                 ConvertWideCharToMultiByte(protocol_->GetCodepage(), data, *str);\r
 231 \r
 232                 socket_->async_write_some(boost::asio::buffer(*str), std::bind(&connection::handle_write, shared_from_this(), str, std::placeholders::_1, std::placeholders::_2));\r
 233         }\r
 234 };\r
 235 \r
 236 struct AsyncEventServer::implementation\r
 237 {\r
 238         boost::asio::io_service                                 service_;\r
 239         tcp::acceptor                                                   acceptor_;\r
 240         spl::shared_ptr<IProtocolStrategy>              protocol_;\r
 241         spl::shared_ptr<connection_set>                 connection_set_;\r
 242         boost::thread                                                   thread_;\r
 243 \r
 244         implementation(const spl::shared_ptr<IProtocolStrategy>& protocol, unsigned short port)\r
 245                 : acceptor_(service_, tcp::endpoint(tcp::v4(), port))\r
 246                 , protocol_(protocol)\r
 247                 , thread_(std::bind(&boost::asio::io_service::run, &service_))\r
 248         {\r
 249                 start_accept();\r
 250         }\r
 251 \r
 252         ~implementation()\r
 253         {\r
 254                 acceptor_.close();\r
 255 \r
 256                 service_.post([=]\r
 257                 {\r
 258                         BOOST_FOREACH(auto& connection, *connection_set_)\r
 259                                 connection->stop();\r
 260                 });\r
 261 \r
 262                 thread_.join();\r
 263         }\r
 264                 \r
 265     void start_accept() \r
 266         {\r
 267                 spl::shared_ptr<tcp::socket> socket(new tcp::socket(service_));\r
 268                 acceptor_.async_accept(*socket, std::bind(&implementation::handle_accept, this, socket, std::placeholders::_1));\r
 269     }\r
 270 \r
 271         void handle_accept(const spl::shared_ptr<tcp::socket>& socket, const boost::system::error_code& error) \r
 272         {\r
 273                 if (!acceptor_.is_open())\r
 274                         return;\r
 275                 \r
 276         if (!error)             \r
 277                         connection_set_->insert(connection::create(socket, protocol_, connection_set_));\r
 278 \r
 279                 start_accept();\r
 280     }\r
 281 };\r
 282 \r
 283 AsyncEventServer::AsyncEventServer(const spl::shared_ptr<IProtocolStrategy>& protocol, unsigned short port) : impl_(new implementation(protocol, port)){}\r
 284 AsyncEventServer::~AsyncEventServer(){}\r
 285 }}