server/Server.cpp

   1 //
   2 // Server.cpp
   3 //
   4
   5 #include "Server.hpp"
   6 #include "version.hpp"
   7 #include <algorithm>
   8 #include <boost/make_shared.hpp>
   9 #include <boost/foreach.hpp>
  10 #include "../common/Logger.hpp"
  11 #include "../common/network/Command.hpp"
  12 #include "../common/network/Utils.hpp"
  13
  14 namespace network {
  15
  16     Server::Server(uint16_t port) :
  17             endpoint_(tcp::v4(), port),
  18             acceptor_(io_service_, endpoint_),
  19             socket_udp_(io_service_, udp::endpoint(udp::v4(), port)),
  20             udp_packet_count_(0),
  21             max_total_read_average_(5000),
  22             max_session_read_average_(600),
  23             min_session_read_average_(100),
  24             session_read_average_(200)
  25     {
  26     }
  27
  28     void Server::Start(CallbackFuncPtr callback)
  29     {
  30         callback_ = std::make_shared<CallbackFunc>(
  31                 [&](network::Command c){
  32
  33             // ログアウト
  34             if (c.header() == network::header::FatalConnectionError) {
  35                 // 受信制限量を更新
  36                 /*
  37                 auto new_average = GetSessionReadAverageLimit();
  38                 if (session_read_average_ != new_average) {
  39                     session_read_average_ = new_average;
  40                     SendAll(network::ClientReceiveWriteAverageLimitUpdate(session_read_average_));
  41                 }
  42                 */
  43             }
  44
  45             // 通信量制限を越えていた場合、強制的に切断
  46      //       else if (auto session = c.session().lock()) {
  47      //           if (session->GetReadByteAverage() > session_read_average_) {
  48                                         //Logger::Info(_T("Banished session: %d"), session->id());
  49      //               session->Close();
  50      //           }
  51      //       }
  52
  53             if (callback) {
  54                 (*callback)(c);
  55             }
  56         });
  57
  58         {
  59         auto new_session = boost::make_shared<ServerSession>(io_service_);
  60         acceptor_.async_accept(new_session->tcp_socket(),
  61                               boost::bind(&Server::ReceiveSession, this, new_session, boost::asio::placeholders::error));
  62         }
  63
  64         {
  65             socket_udp_.async_receive_from(
  66                 boost::asio::buffer(receive_buf_udp_, UDP_MAX_RECEIVE_LENGTH), sender_endpoint_,
  67                 boost::bind(&Server::ReceiveUDP, this,
  68                   boost::asio::placeholders::error,
  69                   boost::asio::placeholders::bytes_transferred));
  70         }
  71
  72         boost::asio::io_service::work work(io_service_);
  73         io_service_.run();
  74     }
  75
  76     void Server::Stop()
  77     {
  78         io_service_.stop();
  79         Logger::Info("stop server");
  80     }
  81     void Server::Stop(int innterrupt_type)
  82     {
  83         io_service_.stop();
  84         Logger::Info(_T("stop server innterrupt_type=%d"),innterrupt_type);
  85     }
  86
  87         int Server::GetUserCount() const
  88         {
  89                 auto count = std::count_if(sessions_.begin(), sessions_.end(),
  90                         [](const SessionWeakPtr& s){ return !s.expired() && s.lock()->online(); });
  91
  92                 return count;
  93         }
  94
  95         std::string Server::GetStatusJSON() const
  96         {
  97                 auto msg = (
  98                                         boost::format("{\"ver\":%d.%d.%d,\"cnt\":%d}")
  99                                                 % MMO_VERSION_MAJOR % MMO_VERSION_MINOR % MMO_VERSION_REVISION %
 100                                                 GetUserCount()
 101                                         ).str();
 102
 103                 return msg;
 104         }
 105
 106     bool Server::Empty() const
 107     {
 108         return GetUserCount() == 0;
 109     }
 110
 111     void Server::ReceiveSession(const SessionPtr& session, const boost::system::error_code& error)
 112     {
 113         if (session_read_average_ > min_session_read_average_) {
 114             session->set_on_receive(callback_);
 115             session->Start();
 116             sessions_.push_back(SessionWeakPtr(session));
 117
 118             // クライアント情報を要求
 119             session->Send(ClientRequestedClientInfo());
 120
 121             // 受信制限量を更新
 122             auto new_average = GetSessionReadAverageLimit();
 123             session->Send(network::ClientReceiveWriteAverageLimitUpdate(session_read_average_));
 124
 125             if (session_read_average_ != new_average) {
 126                 session_read_average_ = new_average;
 127                 SendOthers(network::ClientReceiveWriteAverageLimitUpdate(session_read_average_),
 128                         session);
 129             }
 130
 131         } else {
 132             Logger::Info("Refuse Session");
 133             session->SyncSend(ClientReceiveServerCrowdedError());
 134             session->Close();
 135         }
 136
 137         auto new_session = boost::make_shared<ServerSession>(io_service_);
 138          acceptor_.async_accept(new_session->tcp_socket(),
 139                  boost::bind(&Server::ReceiveSession, this, new_session, boost::asio::placeholders::error));
 140
 141                 RefreshSession();
 142     }
 143
 144         void Server::RefreshSession()
 145         {
 146                 // 使用済のセッションのポインタを破棄
 147         auto it = std::remove_if(sessions_.begin(), sessions_.end(),
 148                 [](const SessionWeakPtr& ptr){
 149             return ptr.expired();
 150         });
 151         sessions_.erase(it, sessions_.end());
 152                 Logger::Info("Active connection: %d", GetUserCount());
 153         }
 154
 155     void Server::SendAll(const Command& command)
 156     {
 157         BOOST_FOREACH(SessionWeakPtr& ptr, sessions_) {
 158             if (auto session = ptr.lock()) {
 159                 session->Send(command);
 160             }
 161         }
 162     }
 163
 164     void Server::SendOthers(const Command& command, SessionWeakPtr self_ptr)
 165     {
 166         BOOST_FOREACH(SessionWeakPtr& ptr, sessions_) {
 167             if (auto session = ptr.lock()) {
 168                 if (auto self = self_ptr.lock()) {
 169                     if (*session != *self) {
 170                         session->Send(command);
 171                     }
 172                 }
 173             }
 174         }
 175     }
 176
 177     void Server::SendUDPTestPacket(const std::string& ip_address, uint16_t port)
 178     {
 179         using boost::asio::ip::udp;
 180
 181         std::stringstream port_str;
 182         port_str << (int)port;
 183
 184         udp::resolver resolver(io_service_);
 185         udp::resolver::query query(udp::v4(), ip_address.c_str(), port_str.str().c_str());
 186         udp::resolver::iterator iterator = resolver.resolve(query);
 187
 188         static char request[] = "MMO UDP Test Packet";
 189         for (int i = 0; i < UDP_TEST_PACKET_TIME; i++) {
 190
 191             io_service_.post(boost::bind(&Server::DoWriteUDP, this, request, *iterator));
 192         }
 193     }
 194
 195     void Server::SendUDP(const std::string& message, const boost::asio::ip::udp::endpoint endpoint)
 196     {
 197                 io_service_.post(boost::bind(&Server::DoWriteUDP, this, message, endpoint));
 198     }
 199
 200     void Server::ReceiveUDP(const boost::system::error_code& error, size_t bytes_recvd)
 201     {
 202         if (bytes_recvd > 0) {
 203             std::string buffer(receive_buf_udp_, bytes_recvd);
 204             FetchUDP(buffer, sender_endpoint_);
 205         }
 206         if (!error) {
 207           socket_udp_.async_receive_from(
 208               boost::asio::buffer(receive_buf_udp_, UDP_MAX_RECEIVE_LENGTH), sender_endpoint_,
 209               boost::bind(&Server::ReceiveUDP, this,
 210                 boost::asio::placeholders::error,
 211                 boost::asio::placeholders::bytes_transferred));
 212         } else {
 213             Logger::Error("%s", error.message());
 214         }
 215     }
 216
 217     void Server::DoWriteUDP(const std::string& msg, const udp::endpoint& endpoint)
 218     {
 219         boost::shared_ptr<std::string> s =
 220               boost::make_shared<std::string>(msg.data(), msg.size());
 221
 222         socket_udp_.async_send_to(
 223             boost::asio::buffer(s->data(), s->size()), endpoint,
 224             boost::bind(&Server::WriteUDP, this,
 225               boost::asio::placeholders::error, s));
 226     }
 227
 228     void Server::WriteUDP(const boost::system::error_code& error, boost::shared_ptr<std::string> holder)
 229     {
 230 //        if (!error) {
 231 //            if (!send_queue_.empty()) {
 232 //                  send_queue_.pop();
 233 //                  if (!send_queue_.empty())
 234 //                  {
 235 //                    boost::asio::async_write(socket_tcp_,
 236 //                        boost::asio::buffer(send_queue_.front().data(),
 237 //                          send_queue_.front().size()),
 238 //                        boost::bind(&Session::WriteTCP, this,
 239 //                          boost::asio::placeholders::error));
 240 //                  }
 241 //            }
 242 //        } else {
 243 //            FatalError();
 244 //        }
 245     }
 246
 247     void Server::FetchUDP(const std::string& buffer, const boost::asio::ip::udp::endpoint endpoint)
 248     {
 249         unsigned char header;
 250                 uint32_t user_id;
 251         unsigned char count;
 252         std::string body;
 253         SessionPtr session;
 254
 255                 size_t readed = 0;
 256         if (buffer.size() > network::Utils::Deserialize(buffer, &header)) {
 257                         readed = network::Utils::Deserialize(buffer, &user_id, &count);
 258                 }
 259
 260         if (readed < buffer.size()) {
 261             body = buffer.substr(readed);
 262         }
 263
 264         if (callback_) {
 265                         (*callback_)(Command(static_cast<network::header::CommandHeader>(header), body, endpoint));
 266         }
 267     }
 268
 269     void Server::ServerSession::Start()
 270     {
 271         online_ = true;
 272
 273         // Nagleアルゴリズムを無効化
 274         socket_tcp_.set_option(boost::asio::ip::tcp::no_delay(true));
 275
 276                 // バッファサイズを変更 1MiB
 277                 boost::asio::socket_base::receive_buffer_size option(1048576);
 278                 socket_tcp_.set_option(option);
 279
 280         // IPアドレスを取得
 281         global_ip_ = socket_tcp_.remote_endpoint().address().to_string();
 282
 283         boost::asio::async_read_until(socket_tcp_,
 284             receive_buf_, NETWORK_UTILS_DELIMITOR,
 285             boost::bind(
 286               &ServerSession::ReceiveTCP, shared_from_this(),
 287               boost::asio::placeholders::error));
 288     }
 289
 290     int Server::GetSessionReadAverageLimit()
 291     {
 292         int byte = max_total_read_average_ / (sessions_.size() + 1);
 293         byte = std::min(byte, max_session_read_average_);
 294
 295         return byte;
 296     }
 297
 298     int Server::max_total_read_average() const
 299     {
 300         return max_total_read_average_;
 301     }
 302
 303     int Server::max_session_read_average() const
 304     {
 305         return max_session_read_average_;
 306     }
 307
 308     int Server::min_session_read_average() const
 309     {
 310         return min_session_read_average_;
 311     }
 312
 313     void Server::set_max_total_read_average(int byte)
 314     {
 315         max_total_read_average_ = byte;
 316     }
 317
 318     void Server::set_max_session_read_average(int byte)
 319     {
 320         max_session_read_average_ = byte;
 321     }
 322
 323     void Server::set_min_session_read_average(int byte)
 324     {
 325         min_session_read_average_ = byte;
 326     }
 327
 328 }