一个简单的ACE Reactor框架的使用

一个简单的ACE Reactor框架的使用

#include "iostream"
#include "ace\OS_NS_unistd.h"
#include "ace\INET_Addr.h"
#include "ace\SOCK_Stream.h"
#include "ace\SOCK_Acceptor.h"
#include "ace\Event_Handler.h"
#include "ace\Reactor.h"
#include "ace\Thread_Manager.h"

using namespace std;

class Server : public ACE_Event_Handler
{
public:
Server(ACE_Reactor *reactor);

int Open(const char *ip, int port);

virtual int handle_input(ACE_HANDLE fd = ACE_INVALID_HANDLE);
virtual int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask);

protected:
~Server(){}

private:
ACE_SOCK_Acceptor acceptor;
ACE_Thread_Manager threadManager;
};

class ClientHandle : public ACE_Event_Handler
{
public:
ClientHandle(ACE_HANDLE handle, ACE_Reactor *re);

virtual int handle_input(ACE_HANDLE fd = ACE_INVALID_HANDLE);
virtual int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask);

ACE_SOCK_Stream &GetSockStream(){ return stream; }

protected:
~ClientHandle(){}

private:
ACE_SOCK_Stream stream;
};

ACE_THR_FUNC_RETURN ClientThread(void *param);

int main(int argc, char *argv[])
{
Server *server = new Server(ACE_Reactor::instance());
server->Open("192.168.60.65", 9000);

// 启动消息循环
ACE_Reactor::instance()->run_event_loop();
return 0;
}

Server::Server(ACE_Reactor *reactor)
:ACE_Event_Handler(reactor)
{

}

int Server::Open(const char *ip, int port)
{
ACE_INET_Addr addr(port, ip);
if (acceptor.open(addr) < 0)
{
return -1;
}

// 注册接收客户端连接消息
return reactor()->register_handler(acceptor.get_handle(), this, ACE_Event_Handler::ACCEPT_MASK);
}

int Server::handle_input(ACE_HANDLE fd)
{
if (fd != acceptor.get_handle())
{
return -1;
}

ACE_SOCK_Stream stream;
if (acceptor.accept(stream) < 0)
{
return -1;
}

ClientHandle *clientHandle = new ClientHandle(stream.get_handle(), reactor());
threadManager.spawn(ClientThread, clientHandle);

return 0;
}

int Server::handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask)
{
acceptor.close();
delete this;
return 0;
}

ClientHandle::ClientHandle(ACE_HANDLE handle, ACE_Reactor *re)
{
stream.set_handle(handle);
reactor(re);
}

int ClientHandle::handle_input(ACE_HANDLE fd)
{
if (stream.get_handle() != fd)
{
return -1;
}

char buf[1024] = { 0 };
if (stream.recv(buf, 1024) <= 0)
{
return -1;
}
cout << "Client:" << buf << endl;

return 0;
}

int ClientHandle::handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask)
{
stream.send("bye client", strlen("bye client"));
stream.close();
delete this;
return 0;
}

ACE_THR_FUNC_RETURN ClientThread(void *param)
{
ClientHandle *clientHandle = (ClientHandle *)param;
clientHandle->reactor()->register_handler(clientHandle->GetSockStream().get_handle(), clientHandle, ACE_Event_Handler::READ_MASK);

ACE_OS::sleep(2);
clientHandle->GetSockStream().send("hello client", strlen("hello client"));
ACE_OS::sleep(2);
clientHandle->reactor()->remove_handler(clientHandle->GetSockStream().get_handle(), ACE_Event_Handler::READ_MASK);
ACE_OS::sleep(2);

return 0;
}

#include "iostream"
#include "ace\INET_Addr.h"
#include "ace\SOCK_Stream.h"
#include "ace\SOCK_Connector.h"
#include "ace\Reactor.h"

using namespace std;

class Client : public ACE_Event_Handler
{
public:
Client(ACE_Reactor *reactor);

virtual int handle_input(ACE_HANDLE fd = ACE_INVALID_HANDLE);
virtual int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask);

int Connect(char *ip, int port);
int Send(char *buf, int len);

protected:
~Client(){}

private:
ACE_SOCK_Stream stream;
ACE_SOCK_Connector connector;
};

int main(int argv, char *argc[])
{
Client *client = new Client(ACE_Reactor::instance());
if (client->Connect("192.168.60.65", 9000) < 0)
{
return 0;
}

client->Send("hello server", strlen("hello server"));

// 启动消息循环
ACE_Reactor::instance()->run_event_loop();
return 0;
}

Client::Client(ACE_Reactor *reactor)
:ACE_Event_Handler(reactor)
{

}

int Client::Connect(char *ip, int port)
{
ACE_INET_Addr addr(port, ip);
// 连接服务
if (connector.connect(stream, addr) < 0)
{
return -1;
}
// 注册读事件
reactor()->register_handler(stream.get_handle(), this, ACE_Event_Handler::READ_MASK);

return 0;
}

int Client::Send(char *buf, int len)
{
return stream.send(buf, len);
}

int Client::handle_input(ACE_HANDLE fd)
{
char buf[1024] = { 0 };
if (stream.recv(buf, 1024) <= 0)
{
return -1;
}
cout << "Server:" << buf << endl;

return 0;
}

int Client::handle_close(ACE_HANDLE handle, ACE_Reactor_Mask close_mask)
{
stream.close();
delete this;
return 0;
}