libs/asio/example/cpp03/chat/chat_server.cpp

   1 //
   2 // chat_server.cpp
   3 // ~~~~~~~~~~~~~~~
   4 //
   5 // Copyright (c) 2003-2017 Christopher M. Kohlhoff (chris at kohlhoff dot com)
   6 //
   7 // Distributed under the Boost Software License, Version 1.0. (See accompanying
   8 // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
   9 //
  10
  11 #include <algorithm>
  12 #include <cstdlib>
  13 #include <deque>
  14 #include <iostream>
  15 #include <list>
  16 #include <set>
  17 #include <boost/bind.hpp>
  18 #include <boost/shared_ptr.hpp>
  19 #include <boost/enable_shared_from_this.hpp>
  20 #include <boost/asio.hpp>
  21 #include "chat_message.hpp"
  22
  23 using boost::asio::ip::tcp;
  24
  25 //----------------------------------------------------------------------
  26
  27 typedef std::deque<chat_message> chat_message_queue;
  28
  29 //----------------------------------------------------------------------
  30
  31 class chat_participant
  32 {
  33 public:
  34   virtual ~chat_participant() {}
  35   virtual void deliver(const chat_message& msg) = 0;
  36 };
  37
  38 typedef boost::shared_ptr<chat_participant> chat_participant_ptr;
  39
  40 //----------------------------------------------------------------------
  41
  42 class chat_room
  43 {
  44 public:
  45   void join(chat_participant_ptr participant)
  46   {
  47     participants_.insert(participant);
  48     std::for_each(recent_msgs_.begin(), recent_msgs_.end(),
  49         boost::bind(&chat_participant::deliver, participant, _1));
  50   }
  51
  52   void leave(chat_participant_ptr participant)
  53   {
  54     participants_.erase(participant);
  55   }
  56
  57   void deliver(const chat_message& msg)
  58   {
  59     recent_msgs_.push_back(msg);
  60     while (recent_msgs_.size() > max_recent_msgs)
  61       recent_msgs_.pop_front();
  62
  63     std::for_each(participants_.begin(), participants_.end(),
  64         boost::bind(&chat_participant::deliver, _1, boost::ref(msg)));
  65   }
  66
  67 private:
  68   std::set<chat_participant_ptr> participants_;
  69   enum { max_recent_msgs = 100 };
  70   chat_message_queue recent_msgs_;
  71 };
  72
  73 //----------------------------------------------------------------------
  74
  75 class chat_session
  76   : public chat_participant,
  77     public boost::enable_shared_from_this<chat_session>
  78 {
  79 public:
  80   chat_session(boost::asio::io_service& io_service, chat_room& room)
  81     : socket_(io_service),
  82       room_(room)
  83   {
  84   }
  85
  86   tcp::socket& socket()
  87   {
  88     return socket_;
  89   }
  90
  91   void start()
  92   {
  93     room_.join(shared_from_this());
  94     boost::asio::async_read(socket_,
  95         boost::asio::buffer(read_msg_.data(), chat_message::header_length),
  96         boost::bind(
  97           &chat_session::handle_read_header, shared_from_this(),
  98           boost::asio::placeholders::error));
  99   }
 100
 101   void deliver(const chat_message& msg)
 102   {
 103     bool write_in_progress = !write_msgs_.empty();
 104     write_msgs_.push_back(msg);
 105     if (!write_in_progress)
 106     {
 107       boost::asio::async_write(socket_,
 108           boost::asio::buffer(write_msgs_.front().data(),
 109             write_msgs_.front().length()),
 110           boost::bind(&chat_session::handle_write, shared_from_this(),
 111             boost::asio::placeholders::error));
 112     }
 113   }
 114
 115   void handle_read_header(const boost::system::error_code& error)
 116   {
 117     if (!error && read_msg_.decode_header())
 118     {
 119       boost::asio::async_read(socket_,
 120           boost::asio::buffer(read_msg_.body(), read_msg_.body_length()),
 121           boost::bind(&chat_session::handle_read_body, shared_from_this(),
 122             boost::asio::placeholders::error));
 123     }
 124     else
 125     {
 126       room_.leave(shared_from_this());
 127     }
 128   }
 129
 130   void handle_read_body(const boost::system::error_code& error)
 131   {
 132     if (!error)
 133     {
 134       room_.deliver(read_msg_);
 135       boost::asio::async_read(socket_,
 136           boost::asio::buffer(read_msg_.data(), chat_message::header_length),
 137           boost::bind(&chat_session::handle_read_header, shared_from_this(),
 138             boost::asio::placeholders::error));
 139     }
 140     else
 141     {
 142       room_.leave(shared_from_this());
 143     }
 144   }
 145
 146   void handle_write(const boost::system::error_code& error)
 147   {
 148     if (!error)
 149     {
 150       write_msgs_.pop_front();
 151       if (!write_msgs_.empty())
 152       {
 153         boost::asio::async_write(socket_,
 154             boost::asio::buffer(write_msgs_.front().data(),
 155               write_msgs_.front().length()),
 156             boost::bind(&chat_session::handle_write, shared_from_this(),
 157               boost::asio::placeholders::error));
 158       }
 159     }
 160     else
 161     {
 162       room_.leave(shared_from_this());
 163     }
 164   }
 165
 166 private:
 167   tcp::socket socket_;
 168   chat_room& room_;
 169   chat_message read_msg_;
 170   chat_message_queue write_msgs_;
 171 };
 172
 173 typedef boost::shared_ptr<chat_session> chat_session_ptr;
 174
 175 //----------------------------------------------------------------------
 176
 177 class chat_server
 178 {
 179 public:
 180   chat_server(boost::asio::io_service& io_service,
 181       const tcp::endpoint& endpoint)
 182     : io_service_(io_service),
 183       acceptor_(io_service, endpoint)
 184   {
 185     start_accept();
 186   }
 187
 188   void start_accept()
 189   {
 190     chat_session_ptr new_session(new chat_session(io_service_, room_));
 191     acceptor_.async_accept(new_session->socket(),
 192         boost::bind(&chat_server::handle_accept, this, new_session,
 193           boost::asio::placeholders::error));
 194   }
 195
 196   void handle_accept(chat_session_ptr session,
 197       const boost::system::error_code& error)
 198   {
 199     if (!error)
 200     {
 201       session->start();
 202     }
 203
 204     start_accept();
 205   }
 206
 207 private:
 208   boost::asio::io_service& io_service_;
 209   tcp::acceptor acceptor_;
 210   chat_room room_;
 211 };
 212
 213 typedef boost::shared_ptr<chat_server> chat_server_ptr;
 214 typedef std::list<chat_server_ptr> chat_server_list;
 215
 216 //----------------------------------------------------------------------
 217
 218 int main(int argc, char* argv[])
 219 {
 220   try
 221   {
 222     if (argc < 2)
 223     {
 224       std::cerr << "Usage: chat_server <port> [<port> ...]\n";
 225       return 1;
 226     }
 227
 228     boost::asio::io_service io_service;
 229
 230     chat_server_list servers;
 231     for (int i = 1; i < argc; ++i)
 232     {
 233       using namespace std; // For atoi.
 234       tcp::endpoint endpoint(tcp::v4(), atoi(argv[i]));
 235       chat_server_ptr server(new chat_server(io_service, endpoint));
 236       servers.push_back(server);
 237     }
 238
 239     io_service.run();
 240   }
 241   catch (std::exception& e)
 242   {
 243     std::cerr << "Exception: " << e.what() << "\n";
 244   }
 245
 246   return 0;
 247 }