ACE Reactor的Echo Server

1 /* ***********************************************************************
2 *@file:echo.cpp
3 *@author:dennis
4 *@revise:dennis<killme2008@gmail.com> http://www.blogjava.net/killme2008
5 *相对完整的echoserver，可以接受多个客户端连接，并且可以通过键入quit正常关闭
6
7 *********************************************************************** */
8
9 #ifdef_DEBUG
10 #pragmacomment(lib, " aced.lib " )
11 #else
12 #pragmacomment(lib, " ace.lib " )
13 #endif
14
15 #include " ace/Reactor.h "
16 #include " ace/SOCK_Acceptor.h "
17 #include " ace/os.h "
18 #include " ace/Log_Msg.h "
19 #include " ace/inet_addr.h "
20 #include " ace/Thread_Manager.h "
21 #include < iostream >
22 #include < string >
23
24 #define PORT_NO 8080
25 typedefACE_SOCK_AcceptorAcceptor;
26 // forwarddeclaration
27 class Echo_Handler;
28
29 class Echo_Handler: public ACE_Event_Handler
30 {
31 public :
32 // construcor
33 Echo_Handler()
34 {
35 }
36 virtual ~ Echo_Handler()
37 {
38 }
39 // Calledbacktohandleanyinputreceived
40 int handle_input(ACE_HANDLE)
41 {
42 // receivethedata
43 ssize_trecvBytes = peer().recv(data, 12 );
44 if (recvBytes <= 0 )
45 {
46 ACE_DEBUG((LM_DEBUG, " %s/n " , " 客户端断开连接 " ));
47 return - 1 ;
48 }
49 data[recvBytes] = 0 ;
50
51 ACE_DEBUG((LM_DEBUG,data));
52
53
54 if (ACE_OS::strcmp(data, " q " ) == 0 )
55 {
56 ACE_DEBUG((LM_DEBUG, " 客户端退出 " ));
57 peer().close();
58 return - 1 ;
59 }
60 peer().send_n(data,recvBytes);
61 // dosomethingwiththeinputreceived.
62 //
63 // keepyourselfregisterdwiththereator
64 return 0 ;
65 }
66
67 int handle_close(ACE_HANDLEh,ACE_Reactor_Maskm)
68 {
69 delete this ;
70 return 0 ;
71 }
72
73 // Usedbythereactortodeterminetheunderlyinghandle
74 ACE_HANDLEget_handle() const
75 {
76 return this -> peer_.get_handle();
77 }
78
79 // Returnsareferencetotheunderlyingstream.
80 ACE_SOCK_Stream & peer()
81 {
82 return this -> peer_;
83 }
84
85 private :
86 ACE_SOCK_Streampeer_;
87 char data[ 12 ];
88 };
89
90 class Echo_Accept_Handler: public ACE_Event_Handler
91 {
92 public :
93 // Constructor
94 Echo_Accept_Handler(ACE_Addr & addr)
95 {
96 this -> open(addr);
97 }
98 virtual ~ Echo_Accept_Handler(){}
99 // Openthepeer_acceptorsoitstartsto"listen"
100 // forincomingclients
101 int open(ACE_Addr & addr)
102 {
103 if (peer_acceptor.open(addr) ==- 1 )
104 ACE_ERROR_RETURN((LM_ERROR, " 启动服务器错误/n " ), 1 );
105 return 0 ;
106 }
107
108 // Overloadthehandleinputmethod
109 int handle_input(ACE_HANDLEhandle)
110 {
111 // Clienthasrequestedconnectiontoserver.
112 // Createahandlertohandletheconnection
113 Echo_Handler * eh;
114 ACE_NEW_RETURN(eh,Echo_Handler, - 1 );
115 ACE_INET_Addrcliaddr;
116 // Accepttheconnection"into"theEventHandler
117 if ( this -> peer_acceptor.accept(eh -> peer(), // stream
118 & cliaddr, // remoteaddress
119 0 , // timeout
120 1 ) == - 1 ) // restartifinterrupted
121 ACE_DEBUG((LM_ERROR, " Errorinconnection/n " ));
122
123 ACE_DEBUG((LM_DEBUG, " 连接已经建立,来自%s/n " ,cliaddr.get_host_addr()));
124
125 // Registertheinputeventhandlerforreading
126 ACE_Reactor::instance() -> register_handler(eh,ACE_Event_Handler::READ_MASK);
127 const char * msg = " 按q键使服务安全退出/r/n " ;
128 eh -> peer().send_n(msg,strlen(msg) + 1 );
129 return 0 ;
130 }
131
132 // Usedbythereactortodeterminetheunderlyinghandle
133 ACE_HANDLEget_handle( void ) const
134 {
135 return this -> peer_acceptor.get_handle();
136 }
137 int handle_close(ACE_HANDLEh,ACE_Reactor_Maskm){
138 peer_acceptor.close();
139 delete this ;
140 return 0 ;
141 }
142
143 private :
144 Acceptorpeer_acceptor;
145 };
146 class Quit_Handler: public ACE_Event_Handler
147 {
148 public :
149 Quit_Handler(ACE_Reactor * r):ACE_Event_Handler(r){}
150 virtual int handle_exception(ACE_HANDLE)
151 {
152 ACE_DEBUG((LM_DEBUG, " 停止服务器中/n " ));
153 reactor() -> end_reactor_event_loop();
154 return - 1 ;
155 }
156 int handle_close(ACE_HANDLEh,ACE_Reactor_Maskm)
157 {
158 delete this ;
159 return 0 ;
160 }
161 virtual ~ Quit_Handler(){}
162 };
163 static ACE_THR_FUNC_RETURNrun_events( void * arg);
164 static ACE_THR_FUNC_RETURNcontroller( void * arg);
165 int ACE_TMAIN( int argc, char * argv[])
166 {
167
168 ACE_Reactor * reactor = ACE_Reactor::instance();
169 if (ACE_Thread_Manager::instance() -> spawn(run_events,reactor,THR_DETACHED | THR_SCOPE_SYSTEM) ==- 1 )
170 return 1 ;
171 if (ACE_Thread_Manager::instance() -> spawn(controller,THR_DETACHED | THR_SCOPE_SYSTEM) ==- 1 )
172 return 1 ;
173 return ACE_Thread_Manager::instance() -> wait();
174 }
175
176 static ACE_THR_FUNC_RETURNrun_events( void * arg)
177 {
178 ACE_Reactor * reactor = ACE_static_cast(ACE_Reactor * ,arg);
179 ACE_INET_Addraddr(PORT_NO);
180
181 Echo_Accept_Handler * eh = 0 ;
182 ACE_NEW_RETURN(eh,Echo_Accept_Handler(addr), 1 );
183
184 ACE_Reactor::instance() -> owner(ACE_OS::thr_self());
185 reactor -> register_handler(eh,ACE_Event_Handler::ACCEPT_MASK);
186 ACE_Reactor::instance() -> run_reactor_event_loop();
187 return 0 ;
188 }
189 static ACE_THR_FUNC_RETURNcontroller( void * arg)
190 {
191 ACE_Reactor * reactor = ACE_static_cast(ACE_Reactor * ,arg);
192 Quit_Handler * quit_handler = 0 ;
193 ACE_NEW_RETURN(quit_handler,Quit_Handler(reactor), 1 );
194 for (;;)
195 {
196 std:: string line;
197 std::getline(std::cin,line, ' /n ' );
198 if (line == " quit " ){
199 ACE_DEBUG((LM_DEBUG, " 请求停止服务器/n " ));
200 reactor -> notify(quit_handler);
201 break ;
202 }
203 }
204 return 0 ;
205 }
206