学步园

    上一篇《ACE通用服务端框架》为我们展示了ACE接受器（Acceptor）和连接器（Connector）设计模式的高度集成度，本篇延续上一篇的思路，完成通用客户端的设计和编码。

    由于ACE充分发挥了C++模板技术的优势，是的Connector模式与Acceptor模式在编码上是如此接近。

    好了，开始我们的编码旅程吧。同样的，本框架部分代码来自《The ACE Programmers Guide》7.6 Using the Acceptor-Connector Framework章节和《ACE自适配通信环境中文技术文档-中篇：ACE程序员教程》第七章的思路。

1、main.cpp

// Base_Client.cpp : 定义控制台应用程序的入口点。<br /> //</p> <p>#include <ace/OS.h><br /> #include <ace/Log_Msg.h><br /> #include <ace/streams.h></p> <p>#include <ace/Reactor.h><br /> #include <ace/TP_Reactor.h><br /> #include <ace/Select_Reactor.h><br /> #include <ace/INET_Addr.h><br /> #include <ace/Signal.h></p> <p>#include "Global_Define.h"</p> <p>int ACE_TMAIN (int, ACE_TCHAR *[])<br /> {<br /> ACE_TP_Reactor *tp_reactor = new ACE_TP_Reactor;<br /> ACE_Reactor *reactor = new ACE_Reactor(tp_reactor, 1);<br /> ACE_Reactor::instance(reactor, 1);</p> <p> //ACE_TP_Reactor tp_reactor;<br /> //ACE_Reactor reactor(&tp_reactor, 0);<br /> //ACE_Reactor::instance(&reactor, 0);</p> <p> //ACE_Select_Reactor select_reactor;<br /> //ACE_Reactor reactor(&select_reactor, 0);<br /> //ACE_Reactor::instance(&reactor, 0);</p> <p> GlobalValue g_val;<br /> if (g_val.load_config(CONFIG_FILE) == -1)<br /> ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT ("(%P|%t) Load config file fail!")), 1);</p> <p> /****************************************************************<br /> * TODO 如果需要，在此处注册信号<br /> *****************************************************************/</p> <p> ScheduleHandler sh;<br /> // 注册信号，处理方法：ScheduleHandler::handle_signal()<br /> ACE_Sig_Set sig_set;<br /> sig_set.sig_add (SIGINT);<br /> sig_set.sig_add (SIGQUIT);<br /> sig_set.sig_add (SIGTERM);<br /> sig_set.sig_add (SIGSEGV);<br /> ACE_Reactor::instance()->register_handler(sig_set, &sh);</p> <p> // 注册读取配置文件定时器，处理方法：ScheduleHandler::handle_timeout()<br /> ACE_Reactor::instance()->schedule_timer(&sh, (const void *)&g_val, ACE_Time_Value(10), ACE_Time_Value(10));</p> <p> //ACE_INET_Addr port_to_connect(50001, ACE_LOCALHOST);<br /> ACE_INET_Addr port_to_connect(g_val.client_port, g_val.client_host.c_str());<br /> ClientConnector connector;</p> <p> do {<br /> Client21 client;<br /> Client21 *pc = &client;<br /> if (connector.connect(pc, port_to_connect) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT ("(%P|%t) %p. %s/n"), ACE_TEXT ("Connect"), ACE_TEXT("Auto re-connect 5 seconds later")));</p> <p> ACE_OS::sleep(5);<br /> continue;<br /> }</p> <p> ACE_Reactor::instance()->run_reactor_event_loop();</p> <p> client.close();</p> <p> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) Auto re-connect 3 seconds later/n")));<br /> ACE_OS::sleep(3);</p> <p> ACE_Reactor::instance()->reset_reactor_event_loop();<br /> } while(true);</p> <p> ACE_Reactor::instance()->close();</p> <p> ACE_Thread_Manager::instance()->wait(&ACE_Time_Value(10));</p> <p> ACE_OS::exit(0);</p> <p> return (0);<br /> }</p> <p>

2、Client21.cpp

/********************************************************************************************<br /> * huangjf 2009年4月 于宁夏银川<br /> ********************************************************************************************/<br /> #include <ace/OS.h><br /> #include <ace/Log_Msg.h><br /> #include <ace/Reactor.h><br /> #include <ace/Message_Block.h><br /> #include <ace/Synch.h><br /> #include <ace/INET_Addr.h><br /> #include <ace/Svc_Handler.h><br /> #include <ace/SOCK_Stream.h><br /> #include <ace/SOCK_Connector.h><br /> #include <ace/Connector.h><br /> #include <ace/Reactor.h><br /> #include <ace/Reactor_Notification_Strategy.h></p> <p>#include "Global_Define.h"</p> <p>/*记录最近一次接入的客户端ACE_SOCK_STREAM，用于控制一次只能允许一个客户端接入的情况*/<br /> //ACE_SOCK_STREAM * last_peer = NULL;</p> <p>Client21::Client21(void) : active_(false) /*方式2*/ /*notifier_(0, this, ACE_Event_Handler::WRITE_MASK)*/<br /> {<br /> this->active_ = false;</p> <p> this->msg_queue()->high_water_mark(SEND_HIGHT_MARK);</p> <p> /*方式2*/<br /> //this->notifier_.reactor(this->reactor());<br /> //this->msg_queue()->notification_strategy(&this->notifier_);</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Client21 start/n")));<br /> return;<br /> }</p> <p>Client21::~Client21(void)<br /> {<br /> this->active_ = false;</p> <p> this->msg_queue()->close();</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Client21 stop/n")));<br /> return;<br /> }</p> <p>// 在连接建立后被自动回调<br /> int Client21::open(void *p)<br /> {<br /> if (super::open(p) == -1) return -1;</p> <p> ACE_TCHAR peer_name[MAXHOSTNAMELEN];<br /> ACE_INET_Addr peer_addr;<br /> if (this->peer().get_remote_addr(peer_addr) == 0 && peer_addr.addr_to_string (peer_name, MAXHOSTNAMELEN) == 0)<br /> ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) Connected to %s/n"), peer_name));</p> <p> this->active_ = true;<br /> task_recv.set_active(true);<br /> task_send.set_active(true);</p> <p> task_recv.set_task_send(&task_send);<br /> if (task_recv.activate(THR_NEW_LWP | THR_JOINABLE, TASK_RECV_POOL_SIZE) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) TaskRecv activate failed/n")));<br /> }</p> <p> task_send.set_task_peer((ACE_Task<ACE_MT_SYNCH> *)this);<br /> if (task_send.activate(THR_NEW_LWP | THR_JOINABLE, TASK_SEND_POOL_SIZE) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) TaskSend activate failed/n")));<br /> }</p> <p> /*回调*/<br /> this->start_0();</p> <p> /*开启定时处理线程*/<br /> ACE_Time_Value interval(1); // One seconds<br /> return this->reactor()->schedule_timer(this, 0, ACE_Time_Value::zero, interval);<br /> //return 0;<br /> }</p> <p>// 客户端接收到服务端发送的数据后，回调此函数<br /> int Client21::handle_input(ACE_HANDLE)<br /> {<br /> char buffer[INPUT_SIZE] = {0};<br /> ssize_t recv_cnt = 0;</p> <p>#if (MSG_HEAD_MODE==1) /*接收消息头模式*/</p> <p> /****************************************************************<br /> * TODO 根据实际消息头格式修改解析方法<br /> *****************************************************************/</p> <p> ssize_t msg_len = 0;</p> <p> recv_cnt = this->peer().recv(buffer, MSG_HEAD_SIZE);</p> <p> if (recv_cnt > 0 && recv_cnt == MSG_HEAD_SIZE)<br /> {<br /> /*************************<br /> * TODO 注意这里的4个字节<br /> **************************/<br /> msg_len = SET_BYTE4(buffer[0]&0xFF) + SET_BYTE3(buffer[1]&0xFF) + SET_BYTE2(buffer[2]&0xFF) + SET_BYTE1(buffer[3]&0xFF);<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Message length (%d) %s/n"), msg_len, msg_len<INPUT_SIZE-MSG_HEAD_SIZE?ACE_TEXT(""):ACE_TEXT(": warning!!! out of range")));</p> <p> if (msg_len > 0 && msg_len < INPUT_SIZE-MSG_HEAD_SIZE)<br /> {<br /> if (msg_len == MSG_HEAD_SIZE)<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Recv(%d+%d):[%s]/n"), MSG_HEAD_SIZE, msg_len-MSG_HEAD_SIZE, buffer));</p> <p> ACE_Message_Block *mb = NULL;<br /> ACE_NEW_RETURN(mb, ACE_Message_Block(msg_len+1), -1);<br /> ACE_OS::memcpy(mb->wr_ptr(), buffer, msg_len);<br /> mb->wr_ptr(msg_len);</p> <p> if (task_recv.putq(mb) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT ("(%P|%t) %p; discarding data/n"), ACE_TEXT ("Client21 enqueue to TaskRecv failed")));<br /> mb->release();<br /> return 0;<br /> }</p> <p> return 0;<br /> }<br /> else<br /> {<br /> recv_cnt = this->peer().recv(buffer+MSG_HEAD_SIZE, (msg_len-MSG_HEAD_SIZE));</p> <p> if (recv_cnt > 0 && recv_cnt == (msg_len-MSG_HEAD_SIZE))<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Recv(%d+%d):[%s]/n"), MSG_HEAD_SIZE, msg_len-MSG_HEAD_SIZE, buffer));</p> <p> ACE_Message_Block *mb = NULL;<br /> ACE_NEW_RETURN(mb, ACE_Message_Block(msg_len+1), -1);<br /> ACE_OS::memcpy(mb->wr_ptr(), buffer, msg_len);<br /> mb->wr_ptr(msg_len);</p> <p> if (task_recv.putq(mb) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT ("(%P|%t) %p; discarding data/n"), ACE_TEXT ("Client21 enqueue to TaskRecv failed")));<br /> mb->release();<br /> return 0;<br /> }</p> <p> return 0;<br /> }</p> <p> if (recv_cnt <= 0 || ACE_OS::last_error() != EWOULDBLOCK)<br /> {<br /> this->active_ = false;</p> <p> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) Connection closed1/n")));</p> <p> /*表示客户端已断开*/<br /> return -1;<br /> }<br /> }<br /> }</p> <p> return 0;<br /> }</p> <p> if (recv_cnt <= 0 || ACE_OS::last_error() != EWOULDBLOCK)<br /> {<br /> this->active_ = false;</p> <p> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) Connection closed/n")));</p> <p> /*表示客户端已断开*/<br /> return -1;<br /> }</p> <p>#else /*不接收消息头模式*/</p> <p> recv_cnt = this->peer().recv(buffer, sizeof(buffer));</p> <p> if (recv_cnt > 0)<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Recv(%d):[%s]/n"), recv_cnt, buffer));</p> <p> ACE_Message_Block *mb = NULL;<br /> ACE_NEW_RETURN(mb, ACE_Message_Block(recv_cnt+1), -1);<br /> ACE_OS::memcpy(mb->wr_ptr(), buffer, recv_cnt);<br /> mb->wr_ptr(recv_cnt);</p> <p> if (task_recv.putq(mb) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT ("(%P|%t) %p; discarding data/n"), ACE_TEXT ("Client21 enqueue to TaskRecv failed")));<br /> mb->release();<br /> return 0;<br /> }</p> <p> return 0;<br /> }</p> <p> if (recv_cnt <= 0 || ACE_OS::last_error() != EWOULDBLOCK)<br /> {<br /> this->active_ = false;</p> <p> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) Connection closed/n")));</p> <p> /*表示客户端已断开*/<br /> return -1;<br /> }</p> <p>#endif</p> <p> return 0;<br /> }</p> <p>// 客户端向服务端发送数据时，回调此函数<br /> int Client21::handle_output (ACE_HANDLE)<br /> {<br /> ACE_Message_Block *mb;<br /> ACE_Time_Value nowait (ACE_OS::gettimeofday ());<br /> while (!this->msg_queue()->is_empty())<br /> if (-1 != this->getq(mb, &nowait))<br /> {<br /> ssize_t send_cnt = this->peer().send (mb->rd_ptr(), mb->length());<br /> if (send_cnt == -1)<br /> {<br /> ACE_ERROR ((LM_ERROR, ACE_TEXT ("(%P|%t) %p/n"), ACE_TEXT ("Client21 send")));<br /> }<br /> else<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Send(%d):[%s]/n"), send_cnt, mb->rd_ptr()));<br /> mb->rd_ptr(ACE_static_cast(size_t, send_cnt));<br /> }</p> <p> mb->release ();<br /> }</p> <p> /*方式2*/<br /> /*<br /> if (this->msg_queue()->is_empty())<br /> this->reactor()->cancel_wakeup(this, ACE_Event_Handler::WRITE_MASK);<br /> else<br /> this->reactor()->schedule_wakeup(this, ACE_Event_Handler::WRITE_MASK);<br /> */</p> <p> /*(方法1)*/ this->reactor()->cancel_wakeup(this, ACE_Event_Handler::WRITE_MASK);</p> <p> return 0;<br /> }</p> <p>// 定时处理函数<br /> int Client21::handle_timeout(const ACE_Time_Value ¤t_time, const void *act)<br /> {<br /> if (this->active_ != true) return 0;</p> <p> this->timeout_0();</p> <p> return 0;<br /> }</p> <p>// 服务端断开时，回调此函数<br /> int Client21::handle_close (ACE_HANDLE h, ACE_Reactor_Mask mask)<br /> {<br /> if (mask == ACE_Event_Handler::WRITE_MASK)<br /> return 0;<br /> else<br /> {<br /> ACE_TCHAR peer_name[MAXHOSTNAMELEN];<br /> ACE_INET_Addr peer_addr;<br /> if (this->peer().get_remote_addr(peer_addr) == 0 && peer_addr.addr_to_string (peer_name, MAXHOSTNAMELEN) == 0)<br /> ACE_DEBUG((LM_INFO, ACE_TEXT("(%P|%t) Disconnected %s/n"), peer_name));</p> <p> /*释放队列资源*/<br /> task_send.set_active(false);<br /> task_recv.set_active(false);<br /> this->set_active(false);</p> <p> task_send.msg_queue()->close();<br /> task_recv.msg_queue()->close();<br /> this->msg_queue()->close();</p> <p> task_send.wait();<br /> task_recv.wait();<br /> this->wait();</p> <p> /*回调*/<br /> this->stop_0();</p> <p> /*关闭链路*/<br /> this->peer().close_reader();<br /> this->peer().close_writer();<br /> this->peer().close();</p> <p> return super::handle_close (h, mask);<br /> }<br /> }</p> <p>/**********************************************************************************************************************************/</p> <p>void Client21::set_active(bool b)<br /> {<br /> this->active_ = b;<br /> }</p> <p>bool Client21::get_active()<br /> {<br /> return this->active_;<br /> }</p> <p>/**********************************************************************************************************************************/</p> <p>void Client21::start_0()<br /> {<br /> /*<br /> if (last_peer != NULL)<br /> {<br /> last_peer->close_reader();<br /> last_peer->close_writer();<br /> last_peer->close();<br /> }<br /> last_peer = &this->peer();<br /> */</p> <p> /****************************************************************<br /> * TODO 可以在此实现接入服务端后需要的初始化工作 - （已经实际接入）<br /> *****************************************************************/</p> <p>}</p> <p>void Client21::stop_0()<br /> {<br /> //if (last_peer == &this->peer()) last_peer = NULL;</p> <p> if (this->get_active() == true)<br /> {<br /> /*退出客户端*/<br /> this->reactor()->end_reactor_event_loop();<br /> }</p> <p> /****************************************************************<br /> * TODO 可以在此实现断开服务端后需要执行的工作<br /> *****************************************************************/</p> <p>}</p> <p>void Client21::timeout_0()<br /> {<br /> /****************************************************************<br /> * TODO 实现客户端定时任务，如发送心跳等<br /> *****************************************************************/</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) handle_timeout/n")));</p> <p>}<br />

3、TaskRecv.cpp

/********************************************************************************************<br /> * huangjf 2009年4月 于宁夏银川<br /> ********************************************************************************************/<br /> #include "Global_Define.h"</p> <p>#include <ace/OS.h><br /> #include <ace/Message_Block.h><br /> #include <ace/Log_Msg.h></p> <p>//处理接收队列<br /> int TaskRecv::svc(void)<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskRecv::svc() start/n")));</p> <p> while (this->active_==true)<br /> {<br /> ACE_Message_Block *mb = NULL;<br /> if (this->getq(mb) == -1)<br /> {<br /> continue;<br /> }</p> <p> process_recv(mb);<br /> }</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskRecv::svc() stop/n")));<br /> return 0;<br /> }</p> <p>void TaskRecv::process_recv(ACE_Message_Block *mb)<br /> {<br /> char * data = mb->rd_ptr();<br /> size_t data_size = mb->length();</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Processing recv message: length %d : %s/n"), data_size, data));</p> <p> /****************************************************************<br /> * TODO 处理接收到的对等端数据，数据的起始地址为data,长度为data_size<br /> *****************************************************************/</p> <p> ACE_OS::sleep(3);</p> <p> //如何发送的例子<br /> send_msg(data, data_size);</p> <p> //数据处理结束必需释放内存<br /> mb->release();<br /> }</p> <p>//向对等端发送数据，实际操作是向发送队列插入一条 ACE_Message_Block<br /> //返回 -1:失败 >=0:发送数据长度<br /> int TaskRecv::send_msg(char * _buf, size_t _size)<br /> {<br /> if (_size == 0) return 0;<br /> if (this->active_ != true) return -1;</p> <p> return this->task_send_->send_msg(_buf, _size);<br /> }</p> <p>/**********************************************************************************************************************************/</p> <p>void TaskRecv::set_active(bool b)<br /> {<br /> this->active_ = b;<br /> }</p> <p>bool TaskRecv::get_active()<br /> {<br /> return this->active_;<br /> }</p> <p>void TaskRecv::set_task_send(TaskSend *_task_send)<br /> {<br /> if (!this->task_send_)<br /> {<br /> this->task_send_ = _task_send;<br /> }<br /> }</p> <p>TaskSend * TaskRecv::get_task_send()<br /> {<br /> return this->task_send_;<br /> }<br />

4、TaskSend.cpp

/********************************************************************************************<br /> * huangjf 2009年4月 于宁夏银川<br /> ********************************************************************************************/<br /> #include "Global_Define.h"</p> <p>#include <ace/OS.h><br /> #include <ace/Message_Block.h><br /> #include <ace/Log_Msg.h></p> <p>//处理发送队列<br /> int TaskSend::svc(void)<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskSend::svc() start/n")));</p> <p> while (this->active_==true)<br /> {<br /> ACE_Message_Block *mb = NULL;<br /> if (this->getq(mb) == -1)<br /> {<br /> continue;<br /> }</p> <p> process_send(mb);<br /> }</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskSend::svc() stop/n")));<br /> return 0;<br /> }</p> <p>void TaskSend::process_send(ACE_Message_Block *mb)<br /> {<br /> if (((Client21 *)this->task_peer_)->get_active() != true) return;</p> <p> char * data = mb->rd_ptr();<br /> size_t data_size = mb->length();</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Processing send message: length %d : %s/n"), data_size, data));</p> <p> if (this->task_peer_->putq(mb) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT ("(%P|%t) %p; discarding data/n"), ACE_TEXT ("TaskSend enqueue to Svc_Handler failed")));<br /> mb->release();<br /> }<br /> /*(方法1)*/ this->task_peer_->reactor()->schedule_wakeup(this->task_peer_, ACE_Event_Handler::WRITE_MASK);<br /> }</p> <p>/**********************************************************************************************************************************/</p> <p>void TaskSend::set_active(bool b)<br /> {<br /> this->active_ = b;<br /> }</p> <p>bool TaskSend::get_active()<br /> {<br /> return this->active_;<br /> }</p> <p>void TaskSend::set_task_peer(ACE_Task<ACE_MT_SYNCH> * _task_peer_)<br /> {<br /> if (!this->task_peer_)<br /> {<br /> this->task_peer_ = _task_peer_;<br /> }<br /> }</p> <p>ACE_Task<ACE_MT_SYNCH> * TaskSend::get_task_peer()<br /> {<br /> return this->task_peer_;<br /> }</p> <p>/**********************************************************************************************************************************/</p> <p>// 可以调用此方法向对等端发送数据，实际操作是向发送队列插入一条 ACE_Message_Block<br /> // 返回 -1:失败 >=0:发送数据长度<br /> int TaskSend::send_msg(char * _buf, size_t _size)<br /> {<br /> if (_size==0) return 0;<br /> if (this->active_ != true) return -1;</p> <p> ACE_Message_Block *mb = NULL;<br /> ACE_NEW_RETURN(mb, ACE_Message_Block(_size+1), -1);<br /> ACE_OS::memcpy(mb->wr_ptr(), _buf, _size);<br /> mb->wr_ptr(_size);</p> <p> if (this->putq(mb) == -1)<br /> {<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT ("(%P|%t) %p; discarding data/n"), ACE_TEXT ("TaskSend enqueue failed")));<br /> mb->release();<br /> return -1;<br /> }</p> <p> return _size;<br /> }<br />

5、ScheduleHandler.cpp

/********************************************************************************************<br /> * huangjf 2009年5月 于福州<br /> ********************************************************************************************/</p> <p>#include "ace/OS.h"<br /> #include "ace/Reactor.h"<br /> #include "ace/Timer_Queue.h"<br /> #include "ace/Time_Value.h"<br /> #include "ace/Event_Handler.h"<br /> #include "ace/Log_Msg.h"</p> <p>#include "Global_Define.h"</p> <p>// 定时任务，定时更新配置参数<br /> int ScheduleHandler::handle_timeout(const ACE_Time_Value &c_tv, const void *arg)<br /> {<br /> GlobalValue * _val = (GlobalValue *)arg;</p> <p> _val->load_config(CONFIG_FILE);</p> <p> return 0;<br /> }</p> <p>// 信号处理<br /> int ScheduleHandler::handle_signal(int signum, siginfo_t *, ucontext_t *)<br /> {<br /> ACE_UNUSED_ARG (signum);</p> <p> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) Receive system signal %d./n"), signum));</p> <p> /****************************************************************<br /> * TODO 如果需要，在此处理信号，信号必需已注册<br /> *****************************************************************/</p> <p> ACE_OS::exit(0);</p> <p> return 0;<br /> }<br />

6、Config.cpp

/********************************************************************************************<br /> * huangjf 2009年5月 于福州<br /> ********************************************************************************************/</p> <p>#include "Global_Define.h"</p> <p>#include <ace/OS.h><br /> #include <ace/Configuration.h><br /> #include <ace/Configuration_Import_Export.h></p> <p>// 载入配置文件<br /> int GlobalValue::load_config(const char* config_filename_)<br /> {<br /> ACE_TString str;<br /> long num = 0;</p> <p> ACE_Configuration_Heap config;<br /> if (config.open() == -1)<br /> {<br /> ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("(%P|%t) %p/n"), ACE_TEXT("config.open()")), -1);<br /> }</p> <p> ACE_Ini_ImpExp config_importer(config);<br /> if (config_importer.import_config(config_filename_) == -1)<br /> {<br /> ACE_ERROR_RETURN((LM_ERROR, ACE_TEXT("(%P|%t) %p/n"), ACE_TEXT(config_filename_)), -1);<br /> }</p> <p> ACE_Configuration_Section_Key status_section;<br /> if (config.open_section (config.root_section(), ACE_TEXT("SYSTEM"), 0, status_section) == -1)<br /> {<br /> ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT("(%P|%t) %p/n"), ACE_TEXT ("Can't open [SYSTEM] section")), -1);<br /> }</p> <p> // 服务地址<br /> if (config.get_string_value(status_section, ACE_TEXT("CLIENT_HOST"), str) != -1)<br /> {<br /> if (str.compare(this->client_host) != 0)<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) [SYSTEM]->'CLIENT_HOST' is updated (%s)->(%s)/n"), this->client_host.c_str(), str.c_str()));<br /> this->client_host = str;<br /> }<br /> str.clear();<br /> }<br /> else<br /> {<br /> this->client_host.clear();<br /> this->client_host = ACE_TEXT("127.0.0.1");<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) [SYSTEM]->'CLIENT_HOST' does not exist, default value is (%s)/n"), this->client_host.c_str()));<br /> }</p> <p> // 服务端口<br /> if (config.get_string_value(status_section, ACE_TEXT("CLIENT_PORT"), str) != -1)<br /> {<br /> try<br /> {<br /> num = ACE_OS::strtol(str.c_str(), NULL, 10);<br /> if (num != this->client_port)<br /> {<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) [SYSTEM]->'CLIENT_PORT' is updated (%d)->(%d)/n"), this->client_port, num));<br /> this->client_port = static_cast<u_short> (num);<br /> }<br /> }<br /> catch (...)<br /> {<br /> this->client_port = 50001;<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) [SYSTEM]->'CLIENT_PORT' is catch exception, default value is (%d)/n"), this->client_port));<br /> }<br /> num = 0;<br /> }<br /> else<br /> {<br /> this->client_port = 50001;<br /> ACE_DEBUG((LM_ERROR, ACE_TEXT("(%P|%t) [SYSTEM]->'CLIENT_PORT' does not exist, default value is (%d)/n"), this->client_port));<br /> }</p> <p> /****************************************************************<br /> * TODO 在此增加读取配置文件的参数<br /> *****************************************************************/</p> <p> return 0;<br /> }<br />

7、Global_Define.h

/********************************************************************************************<br /> * huangjf 2009年4月 于宁夏银川<br /> ********************************************************************************************/<br /> #ifndef _Global_Define_h<br /> #define _Global_Define_h</p> <p>#include <ace/Task.h><br /> #include <ace/Synch.h><br /> #include <ace/INET_Addr.h><br /> #include <ace/Svc_Handler.h><br /> #include <ace/SOCK_Stream.h><br /> #include <ace/SOCK_Connector.h><br /> #include <ace/Connector.h><br /> #include <ace/Reactor.h><br /> #include <ace/Reactor_Notification_Strategy.h></p> <p>#define RECV_HIGHT_MARK 8*1024 /*接收队列高水位标*/<br /> #define SEND_HIGHT_MARK 8*1024 /*发送队列高水位标*/<br /> #define INPUT_SIZE 4*1024 /*接收数据缓冲区大小*/<br /> #define TASK_RECV_POOL_SIZE 8 //处理服务器接收到的客户端数据队列的线程池大小<br /> #define TASK_SEND_POOL_SIZE 1 //处理服务器向客户端发送的数据队列的线程池大小</p> <p>#define MSG_HEAD_MODE 0 /*表示是否采用消息头模式，0:表示不使用，1:表示使用*/<br /> #define MSG_HEAD_SIZE 12</p> <p>#define CONFIG_FILE "../config/config.cfg"</p> <p>#define GET_BYTE1(a) ( (a)%0x100 ) /*低1字节*/<br /> #define GET_BYTE2(a) ( ((a)/0x100)%0x100 ) /*低2字节*/<br /> #define GET_BYTE3(a) ( ((a)/0x10000)%0x100 ) /*低3字节*/<br /> #define GET_BYTE4(a) ( ((a)/0x1000000)%0x100 ) /*低4字节*/</p> <p>#define SET_BYTE1(a) ( (a) ) /*低1字节*/<br /> #define SET_BYTE2(a) ( (a)*0x100 ) /*低2字节*/<br /> #define SET_BYTE3(a) ( (a)*0x10000 ) /*低3字节*/<br /> #define SET_BYTE4(a) ( (a)*0x1000000 ) /*低4字节*/</p> <p>/**********************************************************************************************************************************/</p> <p>/********************************************************************************************<br /> * TODO 全局参数类，在此增加全局参数<br /> ********************************************************************************************/<br /> class GlobalValue<br /> {<br /> public:<br /> ACE_TString client_host; /*客户端主机地址*/<br /> u_short client_port; /*客户端监听端口*/</p> <p> int load_config(const char* config_filename_);<br /> };</p> <p>/**********************************************************************************************************************************/</p> <p>/********************************************************************************************<br /> * 采用ACE_Task任务或主动对象处理模式，处理服务器向客户端发送的数据队列<br /> ********************************************************************************************/<br /> class TaskSend: public ACE_Task<ACE_MT_SYNCH><br /> {<br /> public:<br /> TaskSend() : active_(false), task_peer_(NULL)<br /> {<br /> this->active_ = false;<br /> this->msg_queue()->high_water_mark(SEND_HIGHT_MARK);<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskSend start/n")));<br /> }</p> <p> virtual ~TaskSend()<br /> {<br /> this->active_ = false;<br /> this->msg_queue()->close();<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskSend stop/n")));<br /> }</p> <p> virtual int svc(void);</p> <p> void set_active(bool b);<br /> bool get_active();</p> <p> void set_task_peer(ACE_Task<ACE_MT_SYNCH> * _task_peer_);<br /> ACE_Task<ACE_MT_SYNCH> * get_task_peer();</p> <p> int send_msg(char * _buf, size_t _size);</p> <p>private:<br /> bool active_;</p> <p> ACE_Task<ACE_MT_SYNCH> * task_peer_;</p> <p> void process_send(ACE_Message_Block *mb = NULL);<br /> };</p> <p>/********************************************************************************************<br /> * 采用ACE_Task任务或主动对象处理模式，处理服务器接收到的客户端数据队列<br /> ********************************************************************************************/<br /> class TaskRecv: public ACE_Task<ACE_MT_SYNCH><br /> {<br /> public:<br /> TaskRecv() : active_(false), task_send_(NULL)<br /> {<br /> this->active_ = false;<br /> this->msg_queue()->high_water_mark(RECV_HIGHT_MARK);<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskRecv start/n")));<br /> }</p> <p> virtual ~TaskRecv()<br /> {<br /> this->active_ = false;<br /> this->msg_queue()->close();<br /> ACE_DEBUG((LM_DEBUG, ACE_TEXT("(%P|%t) TaskRecv stop/n")));<br /> }</p> <p> virtual int svc(void);</p> <p> void set_active(bool b);<br /> bool get_active();</p> <p> void set_task_send(TaskSend *_task_send);<br /> TaskSend * get_task_send();</p> <p>private:<br /> bool active_;</p> <p> TaskSend *task_send_;</p> <p> void process_recv(ACE_Message_Block *mb = NULL);<br /> int send_msg(char * _buf, size_t _size);<br /> };</p> <p>/********************************************************************************************<br /> * 采用ACE连接器（Connector）连接建立模式，实现双工单路客户端<br /> ********************************************************************************************/<br /> class Client21 : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_MT_SYNCH><br /> {<br /> typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_MT_SYNCH> super;</p> <p>public:<br /> Client21(void);</p> <p> virtual ~Client21(void);</p> <p> // 在连接建立后被自动回调<br /> virtual int open (void * = 0);</p> <p> // Called when input is available from the client.<br /> virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);</p> <p> // Called when output is possible.<br /> virtual int handle_output (ACE_HANDLE fd = ACE_INVALID_HANDLE);</p> <p> // Called when a timer expires.<br /> virtual int handle_timeout (const ACE_Time_Value ¤t_time, const void *act = 0);</p> <p> // Called when this handler is removed from the ACE_Reactor.<br /> virtual int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask);</p> <p> void set_active(bool b);<br /> bool get_active();</p> <p>private:<br /> bool active_;</p> <p> TaskRecv task_recv;<br /> TaskSend task_send;</p> <p> /********************************************************************************************<br /> * 策略类，实现了Strategy模式。如果ACE_Message_Queue拥有一个策略对象，无论何时有ACE_Message_Block对象<br /> * 进入队列，ACE_Message_Queue都会调用该策略对象的notify()方法，把一个通知放入队列，通知的目标是对象的handle_output()方法<br /> ********************************************************************************************/<br /> /*方式2*/ //ACE_Reactor_Notification_Strategy notifier_; </p> <p> /*回调函数，使用者在此实现对应功能*/<br /> void start_0(); /*对等端接入后回调*/<br /> void stop_0(); /*对等端关闭后回调*/<br /> void timeout_0(); /*定时任务*/<br /> };</p> <p>/**********************************************************************************************************************************/</p> <p>/********************************************************************************************<br /> * 应用程序定时任务，及信号处理<br /> ********************************************************************************************/<br /> class ScheduleHandler : public ACE_Event_Handler<br /> {<br /> public:<br /> virtual int handle_timeout(const ACE_Time_Value ¤t_time, const void *act = 0);</p> <p> virtual int handle_signal(int signum, siginfo_t * = 0, ucontext_t * = 0);<br /> };</p> <p>/**********************************************************************************************************************************/</p> <p>typedef ACE_Connector<Client21, ACE_SOCK_CONNECTOR> ClientConnector;</p> <p>#endif