Chinaunix首页 | 论坛 | 博客

大道之行，天下为公litanhua.blog.chinaunix.net

首页　| 　博文目录　| 　关于我

litanhua_

  • 博客访问： 87502
  • 博文数量： 19
  • 博客积分： 487
  • 博客等级： 下士
  • 技术积分： 205
  • 用 户 组： 普通用户
  • 注册时间： 2011-03-18 15:57
文章分类

全部博文（19）

文章存档

2011年（19）

我的朋友
最近访客
推荐博文
相关博文
boost::asio 异步主动连接多个服务器

分类： 系统运维

2011-03-24 11:22:49

ATC.hpp
  1. #include <boost/asio/deadline_timer.hpp>
  2. #include <boost/asio/io_service.hpp>
  3. #include <boost/asio/ip/tcp.hpp>
  4. #include <boost/asio/read_until.hpp>
  5. #include <boost/asio/streambuf.hpp>
  6. #include <boost/asio/write.hpp>
  7. #include <boost/asio.hpp>
  8. #include <boost/bind.hpp>
  9. #include <boost/shared_ptr.hpp>
  10. #include <boost/enable_shared_from_this.hpp>
  11. #include <iostream>
  12. #include <set>
  13. #include <map>
  14. #include "Message.hpp"
  15. #include "SQLite.hpp"

  17. using boost::asio::deadline_timer;
  18. using boost::asio::ip::tcp;
  19. using boost::asio::ip::udp;

  21. using namespace std;
  22. using namespace PWRD;
  23. using namespace PWRD::SQL;
  24. using namespace PWRD::Protocol;

  26. typedef map<string, int> Peers;
  27. typedef map<string, int> Stats;
  28. typedef map<string, string> Files;
  29. typedef map<string, tcp::socket* > Sockets;
  30. #define DEADLINE 10
  31. #define MAXLEN 1024
  32. #define SECONDS 5

  34. namespace PWRD{
  35.     namespace Network{
  36.         class Session{
  37.             public:
  38.                 Session(boost::asio::ip::tcp::socket &_socket, boost::asio::io_service &io_service, const char *path);
  39.                 ~Session();
  40.                 void start();

  42.                 void stop();
  43.                 void delivery(string &ip);

  45.             private:
  46.                 void handle_connect(const boost::system::error_code& ec,
  47.                         tcp::resolver::iterator endpoint_iter);

  49.                 void start_read();

  51.                 void handle_read_header(const boost::system::error_code& ec);

  53.                 void start_delivery();

  55.                 void handle_read_body(const boost::system::error_code& ec);

  57.                 void start_write_heartbeat();

  59.                 void handle_write_heartbeat(const boost::system::error_code& ec);
  60.                 void handle_wait(const boost::system::error_code& ec);
  61.                 void handle_write_config(const boost::system::error_code& ec, Configure &conf);

  63.             private:
  64.                 bool stopped_;
  65.                 tcp::socket &socket_;
  66.                 Header header_;
  67.                 HeartbeatMessage heartbeat_message_;
  68.                 DataMessage packet_message_;
  69.                 ConfigMessage conf_message_;
  70.                 deadline_timer heartbeat_timer_;
  71.                 Callback *callback_;
  72.                 char *input_stream_;

  74.         };

  76.         class SessionPool{
  77.             public:
  78.                 SessionPool();
  79.                 ~SessionPool();
  80.                 void join(string&, Session *);
  81.                 void delivery(string&);
  82.                 void leave(string &ip);

  84.             private:
  85.                 std::map<string, Session *> session_pool_;    
  86.         };
  87.         /************************************************************
  88.          * Usage: Client will connect to every server and it's never
  89.          *     stopped until you shutdown it. Connect action will
  90.          *    start in every DEADLINE seconds
  91.          ************************************************************/
  92.         class Client{
  93.             public:
  94.                 void stop();
  95.                 void start();
  96.                 Client(boost::asio::io_service &_io_service, boost::asio::io_service &_io_service_data, const char *_sqlite);
  97.             protected:
  98.                 void handle_wait_update();
  99.                 void handle_update();
  100.                 void start_connect();
  101.                 void handle_connect(const boost::system::error_code& ec,Peers::iterator it);
  102.                 void handle_deadline();
  103.                 void check_all();
  104.             private:

  106.                 Peers peers_;
  107.                 Sockets sockets_;
  108.                 Stats status_;
  109.                 Files files_;

  111.                 deadline_timer deadline_;
  112.                 boost::asio::io_service &io_service_;
  113.                 const char * sqlite_;
  114.                 bool all_;

  116.                 SessionPool session_pool_;

  118.                 boost::asio::io_service &io_service_data_;

  120.         };
  121.     };
  122. };

 

ATC.cpp

  1. #include "ATC.hpp"
  2. namespace PWRD{
  3.     namespace Network{
  4.         //-------------------------------------------------------------------------

  6.         //Session

  8.         //-------------------------------------------------------------------------

  10.         Session::Session(boost::asio::ip::tcp::socket &_socket, boost::asio::io_service &io_service, const char *path)
  11.             :socket_(_socket),
  12.             heartbeat_timer_(io_service),
  13.             stopped_(false)
  14.         {
  15.             callback_ = new Callback(path);
  16.             input_stream_ = new char[MAXLEN];
  17.         }

  19.         Session::~Session(){
  20.             delete callback_;
  21.             delete input_stream_;    
  22.         }
  23.         void Session::start(){
  24.             start_read();
  25.             start_write_heartbeat();
  26.         }

  28.         void Session::stop(){
  29.             stopped_ = true;
  30.             socket_.close();
  31.             heartbeat_timer_.cancel();
  32.         }

  34.         void Session::start_read(){
  35.             socket_.async_receive(
  36.                     boost::asio::buffer(&header_, HEADERLEN),
  37.                     boost::bind(&Session::handle_read_header, this,
  38.                         boost::asio::placeholders::error)
  39.                     );
  40.         }

  42.         void Session::handle_read_header(const boost::system::error_code& ec){
  43.             if(stopped_)
  44.                 return;    
  45.             if(!ec){
  46.                 if(HEARTBEAT == header_.type){
  47.                     logs.write_log(NORMAL, "info:Receive heartbeat message");    
  48.                     //heartbeat_timer_.expires_from_now(boost::posix_time::seconds(SECONDS));

  50.                     start_read();
  51.                 }    
  52.                 else if(PACKET == header_.type){
  53.                     socket_.async_receive(
  54.                             boost::asio::buffer(input_stream_, header_.length),
  55.                             boost::bind(&Session::handle_read_body, this,
  56.                                 boost::asio::placeholders::error)
  57.                             );
  58.                     logs.write_log(NORMAL, "info:Receive Packet message: %d bytes", header_.length);    
  59.                 }
  60.             }
  61.             else{
  62.                 logs.write_log(NORMAL, "Error on receive: %s", (ec.message()).c_str());    
  63.                 stop();
  64.             }
  65.         }

  67.         void Session::handle_read_body(const boost::system::error_code& ec){
  68.             if(stopped_)
  69.                 return;    
  70.             if(!ec){
  71.                 if(packet_message_.Parse(input_stream_, header_.length)){
  72.                     Packet * packet = packet_message_.pack();
  73.                     logs.write_log(NORMAL, "info:Receive data: %s", (packet->data()).c_str());
  74.                     callback_->Write(packet);            
  75.                 }

  77.                 start_read();
  78.             }
  79.             else{
  80.                 logs.write_log(NORMAL, "Error on receive: %s", (ec.message()).c_str());    
  81.                 stop();
  82.             }
  83.         }
  84.         void Session::start_write_heartbeat(){
  85.             if(stopped_)
  86.                 return ;    
  87.             Header header;
  88.             header.length = 0;
  89.             header.type = HEARTBEAT;
  90.             boost::asio::async_write(socket_, boost::asio::buffer(&header, HEADERLEN),
  91.                     boost::bind(&Session::handle_write_heartbeat, this,
  92.                         boost::asio::placeholders::error    ));
  93.             logs.write_log(NORMAL, "info:Send heartbeat message--->");
  94.         }
  95.         void Session::handle_write_heartbeat(const boost::system::error_code& ec){
  96.             if(stopped_)
  97.                 return ;    

  99.             if(!ec){
  100.                 heartbeat_timer_.expires_from_now(boost::posix_time::seconds(SECONDS));
  101.                 heartbeat_timer_.async_wait(boost::bind(&Session::start_write_heartbeat, this));

  103.             }
  104.             else{
  105.                 logs.write_log(NORMAL, "info:Error on heartbeat: %s ", (ec.message()).c_str());
  106.                 stop();

  108.             }
  109.         }
  110.         void Session::start_delivery(){
  111.             if(stopped_)
  112.                 return ;    
  113.             /* *
  114.              * SQLite append
  115.              * */
  116.             Configure conf;
  117.             conf_message_.set_conf(conf);
  118.             conf_message_.Serialize();
  119.             header_.length = conf_message_.length();
  120.             header_.type = CONFIG;
  121.             socket_.async_send(
  122.                     boost::asio::buffer(&header_,    HEADERLEN),
  123.                     boost::bind(&Session::handle_write_config, this, _1, conf)
  124.                     //boost::asio::placeholders::error, conf)

  126.                 );
  127.         }

  129.         void Session::delivery(string &ip){
  130.             if(stopped_)
  131.                 return ;    
  132.             /* *
  133.              * SQLite append
  134.              * */
  135.             Configure conf;
  136.             conf_message_.set_conf(conf);
  137.             conf_message_.Serialize();
  138.             header_.length = conf_message_.length();
  139.             header_.type = CONFIG;
  140.             socket_.async_send(
  141.                     boost::asio::buffer(&header_,    HEADERLEN),
  142.                     boost::bind(&Session::handle_write_config, this, _1, conf)
  143.                     //boost::asio::placeholders::error, conf)

  145.                 );
  146.         }

  148.         void Session::handle_write_config(const boost::system::error_code& ec, Configure &conf){
  149.             if(stopped_)
  150.                 return ;    

  152.             if(!ec){
  153.                 socket_.async_send(
  154.                         boost::asio::buffer(conf_message_.data(),
  155.                             conf_message_.length()),
  156.                         boost::bind(&Session::handle_wait, this, _1)

  158.                         );
  159.             }
  160.         }

  162.         void Session::handle_wait(const boost::system::error_code &ec){
  163.             return;    
  164.         }
  165.         
  166.         //-------------------------------------------------------------------------

  168.         //SessionPool

  170.         //-------------------------------------------------------------------------

  172.         SessionPool::SessionPool(){
  173.         }

  175.         SessionPool::~SessionPool(){
  176.         
  177.         }

  179.         void SessionPool::join(string &ip, Session *sess){
  180.             session_pool_[ip] = sess;    
  181.         }

  183.         void SessionPool::delivery(string &ip){
  184.             session_pool_[ip]->delivery(ip);    
  185.         }

  187.         void SessionPool::leave(string &ip){
  188.             if(NULL == session_pool_[ip]){
  189.                 delete session_pool_[ip];    
  190.             }
  191.             session_pool_[ip] = NULL;    
  192.         }

  194.         //-------------------------------------------------------------------------

  196.         //Client

  198.         //-------------------------------------------------------------------------

  200.         Client::Client(boost::asio::io_service &_io_service, boost::asio::io_service &_io_service_data, const char *_sqlite)
  201.             :io_service_(_io_service),
  202.             io_service_data_(_io_service_data),
  203.             deadline_(_io_service),
  204.             sqlite_(_sqlite), all_(false){

  206.             }
  207.         void Client::start()
  208.         {
  209.             //IP Table

  211.             if(NULL == sqlite_){
  212.                 logs.write_log(NORMAL, "Sqlite dbname is invalid");
  213.                 return;
  214.             }    
  215.             SQLite::get_instance(sqlite_);
  216.             AddrMap &map_ = SQLite::lookup();
  217.             logs.write_log(NORMAL, "Get %d records", map_.size());
  218.             AddrMap::iterator iter = map_.begin();
  219.             for(; iter != map_.end(); iter++){
  220.                 Vector vect = iter->second;        
  221.                 string ip = vect[1], port = vect[2];
  222.                 status_[ip] = -1;
  223.                 peers_[ip] = std::atoi(port.c_str());
  224.                 sockets_[ip] = new boost::asio::ip::tcp::socket(io_service_);
  225.                 files_[ip] = "/tmp/file";
  226.                 //times_[ip] = boost::asio::deadline_timer(io_service_);

  228.             }

  230.             //start to connect

  232.             start_connect();
  233.             deadline_.async_wait(boost::bind(&Client::handle_deadline, this));

  235.         }
  236.         void Client::stop(){
  237.             deadline_.cancel();
  238.         }

  240.         void Client::handle_wait_update(){

  242.         }
  243.         void Client::handle_update(){

  245.         }
  246.         void Client::start_connect(){
  247.             if(peers_.size() > 0){
  248.                 deadline_.expires_from_now(boost::posix_time::seconds(DEADLINE));
  249.                 for(Peers::iterator it = peers_.begin();
  250.                         it != peers_.end(); it++){

  252.                     if( status_[it->first] < 0){
  253.                         logs.write_log(NORMAL, "Start to connect: %s:%d", (it->first).c_str(), it->second);    

  255.                         sockets_[it->first]->async_connect(
  256.                                 boost::asio::ip::tcp::endpoint(boost::asio::ip::address::from_string(it->first), it->second),
  257.                                 boost::bind(&Client::handle_connect,
  258.                                     this, _1, it));

  260.                     }
  261.                 }
  262.             }
  263.             else{
  264.                 stop();
  265.             }

  267.         }


  270.         void Client::handle_connect(const boost::system::error_code& ec,
  271.                 Peers::iterator it){
  272.             if(!sockets_[it->first]->is_open()){
  273.                 logs.write_log(NORMAL, "Connect time out");    
  274.                 return;
  275.             }
  276.             else if(ec){
  277.                 logs.write_log(NORMAL, "Connect error:%s", (ec.message()).c_str());
  278.                 //sockets_[it->first]->close();

  280.                 return;
  281.             }
  282.             else{
  283.                 logs.write_log(NORMAL, "Connect ok! it was conected with %s:%d", (it->first).c_str(), it->second);    
  284.                 status_[it->first] = 1;
  285.                 Session * ss = new Session(*(sockets_[it->first]), io_service_data_, (files_[it->first]).c_str());
  286.                 session_pool_.join(it->first, ss);
  287.                 ss->start();
  288.             }
  289.         }

  291.         void Client::handle_deadline(){

  293.             check_all();
  294.             if(!all_){
  295.                 logs.write_log(NORMAL, "Timers out, begin to connect other server");
  296.                 start_connect();    
  297.             }
  298.             deadline_.expires_from_now(boost::posix_time::seconds(DEADLINE));
  299.             deadline_.async_wait(boost::bind(&Client::handle_deadline, this));
  300.         }

  302.         void Client::check_all(){
  303.             for(Stats::iterator it = status_.begin(); it != status_.end(); it++){
  304.                 if(it->second < 0){
  305.                     all_ = false;
  306.                     return;    
  307.                 }    
  308.             }    
  309.             all_ = true;
  310.             return;

  312.         }

  314.     };
  315. };
阅读(7122) | 评论(1) | 转发(0) |
0

上一篇：Base64 编码解码

下一篇：低头需要勇气，抬头需要底气

给主人留下些什么吧！~~

litanhua_2011-03-25 11:25:19

回复 | 举报
关于我们 | 关于IT168 | 联系方式 | 广告合作 | 法律声明 | 免费注册

Copyright 2001-2010 ChinaUnix.net All Rights Reserved 北京皓辰网域网络信息技术有限公司. 版权所有

感谢所有关心和支持过ChinaUnix的朋友们

16024965号-6