我们先看一个服务小例子，并比较Reactor框架
#include <iostream> #include "ace/auto_ptr.h" #include "ace/log_msg.h" #include "ace/inet_addr.h" #include "ace/sock_acceptor.h" #include "ace/reactor.h" #include "ace/acceptor.h"
#include "ace/Connector.h" #include "ace/Reactor_Notification_Strategy.h" #include "ace/Select_Reactor.h" #include "ace/Singleton.h"
#include "ace/svc_handler.h"
#include "ace/Message_Block.h" #include "ace/Message_Queue.h" #include "ace/SOCK_Stream.h"
#include "ace/Null_Mutex.h" #include "ace/Null_Condition.h"
using namespace std;
//服务客户 class ClientService:public ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_NULL_SYNCH> { public: int open(void *p); virtual int handle_input(ACE_HANDLE fd=ACE_INVALID_HANDLE); virtual int handle_output(ACE_HANDLE fd=ACE_INVALID_HANDLE); virtual int handle_close(ACE_HANDLE handle,ACE_Reactor_Mask close_mask);
protected: ACE_SOCK_Stream sock_; ACE_Message_Queue<ACE_NULL_SYNCH> output_queue_; }; int ClientService::open(void *p) { if(ACE_Svc_Handler::open(p)==-1) return -1; ACE_TCHAR peer_name[512]; ACE_INET_Addr peer_addr; if(this->peer().get_remote_addr(peer_addr)==0&&peer_addr.addr_to_string(peer_name,512)==0) cout<<" connection from "<<peer_name<<endl; return 0; } int ClientService::handle_input(ACE_HANDLE) { const size_t INPUT_SIZE=4096; char buffer[INPUT_SIZE]; ssize_t recv_cnt,send_cnt; if((recv_cnt=this->peer().recv(buffer,sizeof(buffer)))<=0) { ACE_DEBUG((LM_DEBUG,ACE_TEXT("(%P|%t) connection closed/n"))); return -1; } send_cnt=this->peer().send(buffer,ACE_static_cast(size_t,recv_cnt)); if(send_cnt==recv_cnt) return 0; if(send_cnt==-1&&ACE_OS::last_error()!=EWOULDBLOCK) ACE_ERROR_RETURN((LM_ERROR,ACE_TEXT("%P|%t) %p/n"),ACE_TEXT("send")),0); if(send_cnt==-1) send_cnt=0; ACE_Message_Block *mb; size_t remaining=ACE_static_cast(size_t,(recv_cnt-send_cnt)); ACE_NEW_RETURN(mb,ACE_Message_Block(&buffer[send_cnt],remaining),-1); int output_off=this->msg_queue()->is_empty(); ACE_Time_Value nowait(ACE_OS::gettimeofday()); if(this->putq(mb,&nowait)==-1) { ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%t)%P;discarding data/n"),ACE_TEXT("enqueue failed "))); mb->release(); return 0; } if(output_off) return this->reactor()->register_handler(this,ACE_Event_Handler::WRITE_MASK); return 0;
} int ClientService::handle_output(ACE_HANDLE) { ACE_Message_Block *mb; ACE_Time_Value nowait(ACE_OS::gettimeofday()); while(0==this->getq(mb,&nowait)) { ssize_t send_cnt=this->peer().send(mb->rd_ptr(),mb->length()); if(send_cnt==-1) ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%t)%p/n"),ACE_TEXT("send"))); else mb->rd_ptr(ACE_static_cast(size_t,send_cnt)); if(mb->length()>0) { this->ungetq(mb); break; } mb->release(); } return (this->msg_queue()->is_empty())?-1:0; } int ClientService::handle_close(ACE_HANDLE h,ACE_Reactor_Mask mask) { if(mask==ACE_Event_Handler::WRITE_MASK) return 0; return ACE_Svc_Handler::handle_close(h,mask); }
typedef ACE_Acceptor<ClientService,ACE_SOCK_ACCEPTOR> ClientAcceptor; int main(int argc,char *argv[]) { ACE_INET_Addr port_to_listen("HAStatus"); cout<<"waiting....."<<endl; ClientAcceptor acceptor; if(acceptor.open(port_to_listen)==-1) return 1; cout<<"waiting....."<<endl; ACE_Reactor::instance()->run_reactor_event_loop(); cout<<"waiting....."<<endl; return 0;
} 不同Reactor框架，此框架无需写ClientAcceptor类,即无需设置acceptor对象的ACE_Reactor实例，只需typedef
typedef ACE_Acceptor<ClientService,ACE_SOCK_ACCEPTOR> ClientAcceptor;
我们继续考察ClientService类
class ClientService:public ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_NULL_SYNCH>
ACE_Svc_Handler允许你指定流类型和加锁类型，与ACE_Acceptor一样，ACE_Svc_Handler需要使用流的地址trait,所以我们使用了ACE_SOCK_STREAM宏，使这些代码在支持或不支持模板trait类型的系统上都能编译
之所以要加锁类型，是因为ACE_Svc_Handler是派生自ACE_Task的，后者含有一个ACE_Message_Queue成员，你必须为这个成员提供同步类型。
在此我们不会使用ACE_Task提供的线程能力，但我们要使用继承而得的ACE_Message_Queue成员，所以我们移除了先前例子的ACE_Message_Queue成员。
另外，我们只使用一个线程，所以用ACE_NULL_SYNCH
注意：ACE_Svc_Handler按照我们通常所需的方式实现了get()_handle（）方法，所以此方法也不见了
接着，再看handle_input(ACE_HANDLE)方法，与前一个版本不同的是：
1.我们使用ACE_Svc_Handler::peer()方法是访问底层的ACE_SOCK_Stream
2.我们通过继承的ACE_Task::msg_queue()方法访问继承的ACE_Message_Queue
3.为了把数据块放入队列中，我们可以继承而得的ACE_Task::putq()方法。同样，我们新的handle_output也类似的，但他利用了继承而得的方法
最后，默认的handle_close()方法会移除所有的反应器登记信息，取消所有的定时器，并删除该处理器
ACE_Svc_Handler的使用极大地简化了我们的服务处理器，使得我们能完全专注于需要的解决的问题，而不需要为所有那些连接管理问题而分心
接着，我们再看ACE_Connector
#include <iostream> #include "ace/reactor.h" #include "ace/inet_addr.h" #include "ace/sock_stream.h" #include "ace/sock_connector.h" #include "ace/connector.h" #include "ace/svc_handler.h" #include "ace/reactor_notification_strategy.h" #include <conio.h> using namespace std; typedef ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_NULL_SYNCH> super; class Client:public ACE_Svc_Handler<ACE_SOCK_STREAM,ACE_NULL_SYNCH> {
public: Client():notifier_(0,this,ACE_Event_Handler::WRITE_MASK) { } virtual int open(void *p=0); virtual int handle_input(ACE_HANDLE fd=ACE_INVALID_HANDLE); virtual int handle_output(ACE_HANDLE fd=ACE_INVALID_HANDLE); virtual int handle_timeout(const ACE_Time_Value ¤t_time,const void *act=0); private: enum{ITERATIONS=5}; int iterations_; ACE_Reactor_Notification_Strategy notifier_; }; int Client::open(void *p) { ACE_Time_Value iter_delay(2); if(ACE_Svc_Handler::open(p)==-1) return -1; this->notifier_.reactor(this->reactor()); this->msg_queue()->notification_strategy(this->notifier_); return this->reactor()->schedule_timer(this,0,ACE_Time_Value::zero,iter_delay); } int Client::handle_input(ACE_HANDLE) { char buf[64]; ssize_t recv_cnt=this->peer().recv(buf,sizeof(buf)-1); if(recv_cnt>0) { } if(recv_cnt==0||ACE_OS::last_error()!=EWOULDBLOCK) { this->reactor()->end_reactor_event_loop(); return -1; } return 0; } int Client::handle_timeout(const ACE_Time_Value &,const void *) { if(this->iterations_>=ITERATIONS) { this->peer().close_writer(); return 0; } ACE_Message_Block *mb; char msg[128]; ACE_OS::sprintf(msg,"iteration %d/n",this->iterations_); ACE_NEW_RETURN(mb,ACE_Message_Block(msg),-1); this->putq(mb); return 0; } int Client::handle_output(ACE_HANDLE) { ACE_Message_Block *mb; ACE_Time_Value nowait(ACE_OS::gettimeofday()); while(-1!=this->getq(mb,&nowait)) { ssize_t send_cnt=this->peer().send(mb->rd_ptr(),mb->length()); if(send_cnt==-1) ACE_ERROR((LM_ERROR,ACE_TEXT("(%P|%T)%P/n"),ACE_TEXT("send"))); else mb->rd_ptr(ACE_static_cast(size_t,send_cnt)); if(mb->length()>0) { this->ungetq(mb); break; } mb->release(); } if(this->msg_queue()->is_empty()) this->reactor()->cancel_wakeup(this,ACE_Event_Handler::WRITE_MASK); else this->reactor()->schedule_wakeup(this,ACE_Event_Handler::WRITE_MASK); return 0; } int ACE_TMAIN(int ,ACE_TCHAR *[]) { ACE_INET_Addr port_to_connect("HAStatus",ACE_LOCALHOST); ACE_Connector<Client,ACE_SOCK_CONNECTOR> connector; Client client; Client *pc=&client; if(connector.connect(pc,port_to_connect)==-1) { ACE_ERROR_RETURN((LM_ERROR,ACE_TEXT("%p/n"),ACE_TEXT("connect")),1); } ACE_Reactor::instance()->run_reactor_event_loop(); getch(); return 0; } 从程序中我们看到有一个新类，ACE_Reactor_Notification_Strategy是一个策略类，实现了Strategy模式，它允许你定制另一个类的行为，且无需改变受影响的类
在例子中也可以看到构造的时候先 初始化了notifier_对象，设置正确的ACE_Reactor指针，目的是通过 ACE_Reactor_Notification_Strategy这样的对象使ACE_Message_Queue策略化。如果 ACE_Message_Queue拥有一个策略对象，无论何时有ACE_Message_Block对象进入队列，ACE_Message_Queue 都会调用该策略对象的notify()方法。因为我们已经设置了notifier_,他会在Client的反应器上发出notify()调用，把一个通知 放入队列，通知的目标是我们的client对象的handle_output()方法。要把ACE_Message_Queue的入队操作结合进反应器的 事件循环
handle_timeout方法：如果我们把预定数目的串发给服务器，就是用close_writer()方法关闭我们这一端的TCP/IP socket。
注意，对于我们想要发往服务器的 每一个串，我们都把它插入一个ACE_Message_Block,并把这个块放入队列，这将使消息队列用notifier_对象把通知放入反应器的队 列。当反应器处理该通知时，他会调用我们的handle_output方法，然后我们从队列中取出数据，直到队列变空。
在handle_output方 法中，有cancel_wakeup()和schedule_wakeup(),前者是从这个处理器的反应器登记信息中移除指定的掩码位，而 schedule_wakeup()则增加指定的掩码。也就是说不会造成handle_close()被调用。因此，这一次也无需实现 handle_close(),以专门处理被取消的WRITE掩码，我们复用了ACE_Svc_Handler的默认handle_close()方法， 所以无需编写代码
