server/Server.cpp

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