boost::asio学习之[六]简单聊天程序

#pragma once
#include <cstdio>
#include <cstdlib>
#include <cstring>

namespace ChatTest
{
class chat_message
{
public:
enum {header_length = 4, max_body_length = 1024};
public:
chat_message() : body_length_(0) {}

size_t length() const
{
return header_length + body_length_;
}

void body_length(size_t new_length)
{
body_length_ = new_length;
if (body_length_ > max_body_length)
body_length_ = max_body_length;
}

size_t body_length() const
{
return body_length_;
}

char* data()
{
return data_;
}

const char* data() const
{
return data_;
}

char* body()
{
return data_ + header_length;
}

const char* body() const
{
return data_ + header_length;
}

bool encode_header()
{
char header[header_length+1] = "";
sprintf(header, "%4d", body_length_);
memcpy(data_, header, header_length);
return true;
}

bool decode_header()
{
char header[header_length + 1] = "";
strncat(header, data_, header_length);
body_length_ = atoi(header);
std::cout << body_length_ << std::endl;
if (body_length_ > max_body_length)
{
body_length_ = 0;
return false;
}
return true;
}

private:
char data_[header_length + max_body_length];
size_t body_length_;
};

}

#pragma once
#include <deque>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include "my_chat_message.h"

namespace ChatTest
{
using namespace boost::asio::ip;
typedef std::deque<chat_message> chat_message_queue;

class chat_client
{
public:
chat_client(boost::asio::io_service& io, tcp::resolver::iterator endpoint_iterator)
:io_service_(io), socket_(io)
{
start_connect(endpoint_iterator);
}

void start_connect(tcp::resolver::iterator endpoint_iterator)
{
boost::asio::async_connect(socket_,
endpoint_iterator,
boost::bind(&chat_client::hand_connect, this, boost::asio::placeholders::error));
}

void write(const chat_message& msg)//投入到队列中去
{
io_service_.post(boost::bind(&chat_client::do_write, this, msg));
}

void close()
{
io_service_.post(boost::bind(&chat_client::do_close, this));
}

private:
void hand_connect(const boost::system::error_code& error)
{
if (!error)
{
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg_.data(), chat_message::header_length),
boost::bind(&chat_client::hand_read_header, this, boost::asio::placeholders::error));
}
else
{
//std::cout << "connect error\n" << std::endl;
}
}

void hand_read_header(const boost::system::error_code& error)
{
if (!error && read_msg_.decode_header())
{
std::cout.write(read_msg_.data(), chat_message::header_length);
//std::cout << "\nheader\n" << read_msg_.body_length() << std::endl;
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg_.body(), read_msg_.body_length()/*chat_message::max_body_length长度大于实际长度时会出做*/),
boost::bind(&chat_client::hand_read_body, this, boost::asio::placeholders::error));
}
else
{
//std::cout << "connect error\n" << std::endl;
do_close();
}
}

void hand_read_body(const boost::system::error_code& error)
{
if (!error)
{
std::cout.write(read_msg_.body(), read_msg_.body_length());
std::cout << "\n";
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg_.data(), chat_message::header_length),
boost::bind(&chat_client::hand_read_header, this, boost::asio::placeholders::error));
}
else
{
//std::cout << "connect error\n" << std::endl;
do_close();
}
}

void do_write(const chat_message& msg)
{
bool write_in_progress = !write_msgs_.empty();
write_msgs_.push_back(msg);
if (!write_in_progress)//为空，发送！
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(), write_msgs_.front().length()),
boost::bind(&chat_client::hand_write, this, boost::asio::placeholders::error));
}
}

void hand_write(const boost::system::error_code& error)
{
if (!error)
{
write_msgs_.pop_front();
if (!write_msgs_.empty())//非空，就发送！
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(),write_msgs_.front().length()),
boost::bind(&chat_client::hand_write, this, boost::asio::placeholders::error));
}
}
else
{
do_close();
}
}

void do_close()
{
socket_.close();
}

private:
boost::asio::io_service& io_service_;
tcp::socket socket_;
chat_message read_msg_;
chat_message_queue write_msgs_;
};

void test_cliet()
{
try
{
boost::asio::io_service io;
tcp::resolver resolver(io);
tcp::resolver::query query("localhost", "6688");
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);

chat_client c(io, endpoint_iterator);

boost::thread t(boost::bind(&boost::asio::io_service::run, &io));

char line[chat_message::max_body_length +1];
while (std::cin.getline(line, chat_message::max_body_length+1))
{
chat_message msg;
msg.body_length(strlen(line));
memcpy(msg.body(), line, msg.body_length());
msg.encode_header();
c.write(msg);
}

c.close();
t.join();
}
catch(std::exception& e)
{
std::cerr << e.what() << std::endl;
}
}
}

#pragma once
#include <deque>
#include <set>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/enable_shared_from_this.hpp>
#include "my_chat_message.h"

namespace ChatTest
{
using namespace boost::asio::ip;
typedef std::deque<chat_message> chat_msg_queue;

class chat_participant
{
public:
chat_participant() {}
virtual ~chat_participant() {}

virtual void deliver(const chat_message& msg) = 0;
};

typedef boost::shared_ptr<chat_participant> chat_participant_ptr;

class chat_room
{
public:
chat_room() {}

void join(chat_participant_ptr participant)
{
participants_.insert(participant);
std::for_each(recent_msgs_.begin(), recent_msgs_.end(),
boost::bind(&chat_participant::deliver, participant, _1));
}

void leave(chat_participant_ptr participant)
{
participants_.erase(participant);
}

void deliver(const chat_message& msg)
{
recent_msgs_.push_back(msg);
while (recent_msgs_.size() > max_msgs_size)
{
recent_msgs_.pop_front();
}

std::for_each(participants_.begin(), participants_.end(),
boost::bind(&chat_participant::deliver, _1, boost::ref(msg)));//对每一个参与者发送消息！
}

private:
std::set<chat_participant_ptr> participants_;
enum{max_msgs_size = 100};
chat_msg_queue recent_msgs_;
};

class chat_session
: public chat_participant,
public boost::enable_shared_from_this<chat_session>
{
public:
chat_session(boost::asio::io_service& io, chat_room& room)
:socket_(io), room_(room)
{
}

void start()
{
room_.join(shared_from_this());
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg.data(), chat_message::header_length),
boost::bind(&chat_session::handle_read_header, shared_from_this(), boost::asio::placeholders::error));
}

virtual void deliver(const chat_message& msg)
{
bool write_in_progress = !write_msgs_.empty();
write_msgs_.push_back(msg);
if (!write_in_progress)
{
boost::asio::async_write(socket_,
boost::asio::buffer(msg.data(), msg.length()),
boost::bind(&chat_session::hand_write, shared_from_this(), boost::asio::placeholders::error));
}
}

tcp::socket& socket()
{
return socket_;
}

private:
void hand_write(const boost::system::error_code& error)
{
if (!error)
{
write_msgs_.pop_front();
if (!write_msgs_.empty())
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(), write_msgs_.front().length()),
boost::bind(&chat_session::hand_write, shared_from_this(), boost::asio::placeholders::error));
}
}
else
{
room_.leave(shared_from_this());
}
}

void handle_read_header(const boost::system::error_code& error)
{
if (!error && read_msg.decode_header())
{
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg.body(), read_msg.body_length()),
boost::bind(&chat_session::handle_read_body, shared_from_this(), boost::asio::placeholders::error));
}
else
{
room_.leave(shared_from_this());
}
}

void handle_read_body(const boost::system::error_code& error)
{
if (!error)
{
room_.deliver(read_msg);//读完之后，发送
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg.data(), chat_message::header_length),
boost::bind(&chat_session::handle_read_header, shared_from_this(), boost::asio::placeholders::error));
}
else
{
room_.leave(shared_from_this());
}
}

private:
boost::asio::io_service io_service_;
chat_message read_msg;
chat_msg_queue write_msgs_;
tcp::socket socket_;
chat_room& room_;
};

typedef boost::shared_ptr<chat_session> chat_session_ptr;

class chat_server
{
public:
chat_server(boost::asio::io_service& io, const tcp::endpoint& endpoint)
:io_service_(io), acceptor_(io, endpoint)
{
start_accept();
}

void start_accept()
{
chat_session_ptr new_session(new chat_session(io_service_, room_));
acceptor_.async_accept(new_session->socket(),
boost::bind(&chat_server::hand_accept, this, new_session, boost::asio::placeholders::error));
}
private:

void hand_accept(chat_session_ptr new_session, const boost::system::error_code& error)
{
if (!error)
{
new_session->start();
}
start_accept();
}

private:
boost::asio::io_service& io_service_;
tcp::acceptor acceptor_;
chat_room room_;
};

typedef boost::shared_ptr<chat_server> chat_server_ptr;
typedef std::list<chat_server_ptr> server_list_type;

void test_sever()
{
try
{
boost::asio::io_service io;
server_list_type server_list;

for (int i = 1; i < 3; ++i)
{
tcp::endpoint endpoint(tcp::v4(), 6688);
chat_server_ptr server(new chat_server(io, endpoint));
server_list.push_back(server);
}

io.run();
}
catch (std::exception& e)
{
std::cerr << e.what() << std::endl;
}
}
}

#pragma once
#include <deque>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include "my_chat_message.h"

namespace ChatTest
{
using namespace boost::asio::ip;
typedef std::deque<chat_message> chat_message_queue;

class chat_client
{
public:
chat_client(boost::asio::io_service& io, tcp::resolver::iterator endpoint_iterator)
:io_service_(io), socket_(io)
{
start_connect(endpoint_iterator);
}

void start_connect(tcp::resolver::iterator endpoint_iterator)
{
boost::asio::async_connect(socket_,
endpoint_iterator,
boost::bind(&chat_client::hand_connect, this, boost::asio::placeholders::error));
}

void write(const chat_message& msg)//投入到队列中去
{
io_service_.post(boost::bind(&chat_client::do_write, this, msg));
}

void close()
{
io_service_.post(boost::bind(&chat_client::do_close, this));
}

private:
void hand_connect(const boost::system::error_code& error)
{
if (!error)
{
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg_.data(), chat_message::header_length),
boost::bind(&chat_client::hand_read_header, this, boost::asio::placeholders::error));
}
else
{
//std::cout << "connect error\n" << std::endl;
}
}
/************************************************************************/
/* async_read(socket_, boost::asio::buffer(data_, read_size), handler) ：直到
/* 读到read_size之后才返回，Consider using the async_read function
/*if you need to ensure that the requested amount of data is read
/*before the asynchronous operation completes.
/* async_read_some(socket, boost::asio::buffer(data_, read_max_size) ：
/* The read operation may not read all of the requested number of bytes.
/************************************************************************/
void hand_read_header(const boost::system::error_code& error)
{
if (!error && read_msg_.decode_header())
{
std::cout.write(read_msg_.data(), chat_message::header_length);
std::cout << "\nheader\n" << read_msg_.body_length() << std::endl;
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg_.body(), chat_message::max_body_length/*read_msg_.body_length()长度大于实际长度时会出做*/),
boost::bind(&chat_client::hand_read_body, this, boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred));
}
else
{
//std::cout << "connect error\n" << std::endl;
do_close();
}
}
/************************************************************************/
/* The async_write_some operation may not transmit all of the data to the peer.
/*Consider using the async_write function if you need to ensure that all data is written
/*before the asynchronous operation completes.                                                                     */
/************************************************************************/
void hand_read_body(const boost::system::error_code& error, size_t bytes_read)
{
if (!error)
{
//std::cout.write(read_msg_.body(), read_msg_.body_length());
std::cout.write(read_msg_.body(), bytes_read);
std::cout << bytes_read << " LLLLLLLLL\n";
std::cout << "\n";
boost::asio::async_read(socket_,
boost::asio::buffer(read_msg_.data(), chat_message::header_length),
boost::bind(&chat_client::hand_read_header, this, boost::asio::placeholders::error));
}
else
{
//std::cout << "connect error\n" << std::endl;
do_close();
}
}

void do_write(const chat_message& msg)
{
bool write_in_progress = !write_msgs_.empty();
write_msgs_.push_back(msg);
if (!write_in_progress)//为空，发送！
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(), write_msgs_.front().length()),
boost::bind(&chat_client::hand_write, this, boost::asio::placeholders::error));
}
}

void hand_write(const boost::system::error_code& error)
{
if (!error)
{
write_msgs_.pop_front();
if (!write_msgs_.empty())//非空，就发送！
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(),write_msgs_.front().length()),
boost::bind(&chat_client::hand_write, this, boost::asio::placeholders::error));
}
}
else
{
do_close();
}
}

void do_close()
{
socket_.close();
}

private:
boost::asio::io_service& io_service_;
tcp::socket socket_;
chat_message read_msg_;
chat_message_queue write_msgs_;
};

void test_cliet()
{
try
{
boost::asio::io_service io;
tcp::resolver resolver(io);
tcp::resolver::query query("localhost", "6688");
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);

chat_client c(io, endpoint_iterator);

boost::thread t(boost::bind(&boost::asio::io_service::run, &io));

char line[chat_message::max_body_length +1];
while (std::cin.getline(line, chat_message::max_body_length+1))
{
chat_message msg;
msg.body_length(strlen(line));
memcpy(msg.body(), line, msg.body_length());
msg.encode_header();
c.write(msg);
}

c.close();
t.join();
}
catch(std::exception& e)
{
std::cerr << e.what() << std::endl;
}
}
}