ACE程序员教程-第7章 接受器和连接器：连接建立模式

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

#include ”ace/Thread.h”

//Add our own Reactor singleton

typedef ACE_Singleton Reactor;

//Create an Acceptor

typedef ACE_Acceptor Acceptor;

//Create a Connector

typedef ACE_Connector Connector;

class MyServiceHandler:

public ACE_Svc_Handler

{

public:

//Used by the two threads “globally” to determine their peer stream

static ACE_SOCK_Stream* Peer;

//Thread ID used to identify the threads

ACE_thread_t t_id;

int open(void*)

{

cout<<”Acceptor: received new connection”<

　

//Register with the reactor to remember this handle

Reactor::instance()

->register_handler(this,ACE_Event_Handler::READ_MASK);

　

//Determine the peer stream and record it globally

MyServiceHandler::Peer=&peer();

　

//Spawn new thread to send string every second

ACE_Thread::spawn((ACE_THR_FUNC)send_data,0,THR_NEW_LWP,&t_id);

　

//keep the service handler registered by returning 0 to the

//reactor

return 0;

}

static void* send_data(void*)

{

while(1)

{

cout<<”>>Hello World”<

Peer->send_n(”Hello World”,sizeof(”Hello World”));

　

//Go to sleep for a second before sending again

ACE_OS::sleep(1);

}

　

return 0;

}

int handle_input(ACE_HANDLE)

{

char* data= new char[12];

　

//Check if peer aborted the connection

if(Peer.recv_n(data,12)==0)

{

cout<<”Peer probably aborted connection”);

ACE_Thread::cancel(t_id); //kill sending thread ..

return -1; //de-register from the Reactor.

}

　

//Show what you got..

cout<<”<< %s\n”,data”<

　

//keep yourself registered

return 0;

}

};

//Global stream identifier used by both threads

ACE_SOCK_Stream * MyServiceHandler::Peer=0;

void main_accept()

{

ACE_INET_Addr addr(PORT_NO);

Acceptor myacceptor(addr,Reactor::instance());

while(1)

Reactor::instance()->handle_events();

　

return 0;

}

void main_connect()

{

ACE_INET_Addr addr(PORT_NO,HOSTNAME);

Connector myconnector;

myconnector.connect(my_svc_handler,addr);

while(1)

Reactor::instance()->handle_events();

}

int main(int argc, char* argv[])

{

// Use ACE_Get_Opt to parse and obtain arguments and then call the

// appropriate function for accept or connect.

...

}

这个简单的例子演示怎样联合使用接受器和连接模式来生成服务处理例程，该例程与底层的网络连接建立方法是完全分离的。通过改变相应的设定具体连接器和接受器的模板参数，可以很容易地改用任何其他的底层网络连接建立协议。

class MySvcHandler:

public ACE_Svc_Handler

{

...

}

// = Active object activation method.

virtual int activate (long flags = THR_NEW_LWP,

int n_threads = 1,

int force_active = 0,

long priority = ACE_DEFAULT_THREAD_PRIORITY,

int grp_id = -1,

ACE_Task_Base *task = 0,

ACE_hthread_t thread_handles[] = 0,

void *stack[] = 0,

size_t stack_size[] = 0,

ACE_thread_t thread_names[] = 0);

THR_CANCEL_DISABLE, THR_CANCEL_ENABLE, THR_CANCEL_DEFERRED,

THR_CANCEL_ASYNCHRONOUS, THR_BOUND, THR_NEW_LWP, THR_DETACHED,

THR_SUSPENDED, THR_DAEMON, THR_JOINABLE, THR_SCHED_FIFO,

THR_SCHED_RR, THR_SCHED_DEFAULT

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

class MyServiceHandler; //forward declaration

typedef ACE_Singleton Reactor;

typedef ACE_Acceptor Acceptor;

class MyServiceHandler:

public ACE_Svc_Handler

{

// The two thread names are kept here

ACE_thread_t thread_names[2];

public:

int open(void*)

{

ACE_DEBUG((LM_DEBUG, ”Acceptor: received new connection \n”));

　

//Register with the reactor to remember this handler..

Reactor::instance()

->register_handler(this,ACE_Event_Handler::READ_MASK);

ACE_DEBUG((LM_DEBUG,”Acceptor: ThreadID:(%t) open\n”));

　

//Create two new threads to create and send messages to the

//remote machine.

activate(THR_NEW_LWP,

2, //2 new threads

0, //force active false, if already created don’t try again.

ACE_DEFAULT_THREAD_PRIORITY,//Use default thread priority

-1,

this,//Which ACE_Task object to create? In this case this one.

0,// don’t care about thread handles used

0,// don’t care about where stacks are created

0,//don’t care about stack sizes

thread_names); // keep identifiers in thread_names

　

//keep the service handler registered with the acceptor.

return 0;

}

void send_message1(void)

{

//Send message type 1

ACE_DEBUG((LM_DEBUG,”(%t)Sending message::>>”));

　

//Send the data to the remote peer

ACE_DEBUG((LM_DEBUG,”Sent message1”));

peer().send_n(”Message1”,LENGTH_MSG_1);

} //end send_message1

int send_message2(void)

{

//Send message type 1

ACE_DEBUG((LM_DEBUG,”(%t)Sending message::>>”));

　

//Send the data to the remote peer

ACE_DEBUG((LM_DEBUG,”Sent Message2”));

peer().send_n(”Message2”,LENGTH_MSG_2);

}//end send_message_2

　

int svc(void)

{

ACE_DEBUG( (LM_DEBUG,”(%t) Svc thread \n”));

if(ACE_Thread::self()== thread_names[0])

while(1) send_message1(); //send message 1s forever

else

while(1) send_message2(); //send message 2s forever

　

return 0; // keep the com, piler happy.

}

int handle_input(ACE_HANDLE)

{

ACE_DEBUG((LM_DEBUG,”(%t) handle_input ::”));

char* data= new char[13];

　

//Check if peer aborted the connection

if(peer().recv_n(data,12)==0)

{

printf(”Peer probably aborted connection”);

return -1; //de-register from the Reactor.

}

　

//Show what you got..

ACE_OS::printf(”<< %s\n”,data);

　

//keep yourself registered

return 0;

}

};

int main(int argc, char* argv[])

{

ACE_INET_Addr addr(10101);

ACE_DEBUG((LM_DEBUG,”Thread: (%t) main”));

//Prepare to accept connections

Acceptor myacceptor(addr,Reactor::instance());

// wait for something to happen.

while(1)

Reactor::instance()->handle_events();

　

return 0;

}

　　下面的例子演示怎样使用消息队列的一些功能：

#include ”ace/Reactor.h”

#include ”ace/Svc_Handler.h”

#include ”ace/Acceptor.h”

#include ”ace/Synch.h”

#include ”ace/SOCK_Acceptor.h”

#include ”ace/Thread.h”

#define NETWORK_SPEED 3

class MyServiceHandler; //forward declaration

typedef ACE_Singleton Reactor;

typedef ACE_Acceptor Acceptor;

class MyServiceHandler:

public ACE_Svc_Handler{

// The message sender and creator threads are handled here.

ACE_thread_t thread_names[2];

public:

int open(void*)

{

ACE_DEBUG((LM_DEBUG, ”Acceptor: received new connection \n”));

　

//Register with the reactor to remember this handler..

Reactor::instance()

->register_handler(this,ACE_Event_Handler::READ_MASK);

ACE_DEBUG((LM_DEBUG,”Acceptor: ThreadID:(%t) open\n”));

　

//Create two new threads to create and send messages to the

//remote machine.

activate(THR_NEW_LWP,

2, //2 new threads

0,

ACE_DEFAULT_THREAD_PRIORITY,

-1,

this,

0,

0,

0,

thread_names); // identifiers in thread_handles

　

//keep the service handler registered with the acceptor.

return 0;

}

void send_message(void)

{

//Dequeue the message and send it off

ACE_DEBUG((LM_DEBUG,”(%t)Sending message::>>”));

　

//dequeue the message from the message queue

ACE_Message_Block *mb;

ACE_ASSERT(this->getq(mb)!=-1);

int length=mb->length();

char *data =mb->rd_ptr();

　

//Send the data to the remote peer

ACE_DEBUG((LM_DEBUG,”%s \n”,data,length));

peer().send_n(data,length);

　

//Simulate very SLOW network.

ACE_OS::sleep(NETWORK_SPEED);

　

//release the message block

mb->release();

} //end send_message

int construct_message(void)

{

// A very fast message creation algorithm

// would lead to the need for queuing messages..

// here. These messages are created and then sent

// using the SLOW send_message() routine which is

// running in a different thread so that the message

//construction thread isn’t blocked.

ACE_DEBUG((LM_DEBUG,”(%t)Constructing message::>> ”));

　

// Create a new message to send

ACE_Message_Block *mb;

char *data=”Hello Connector”;

ACE_NEW_RETURN (mb,ACE_Message_Block (16,//Message 16 bytes long

ACE_Message_Block::MB_DATA,//Set header to data

0,//No continuations.

data//The data we want to send

), 0);

mb->wr_ptr(16); //Set the write pointer.

　

// Enqueue the message into the message queue

// we COULD have done a timed wait for enqueuing in case

// someone else holds the lock to the queue so it doesn’t block

//forever..

ACE_ASSERT(this->putq(mb)!=-1);

ACE_DEBUG((LM_DEBUG,”Enqueued msg successfully\n”));

}

int svc(void)

{

ACE_DEBUG( (LM_DEBUG,”(%t) Svc thread \n”));

　

//call the message creator thread

if(ACE_Thread::self()== thread_names[0])

while(1) construct_message(); //create messages forever

else

while(1) send_message(); //send messages forever

　

return 0; // keep the compiler happy.

}

int handle_input(ACE_HANDLE)

{

ACE_DEBUG((LM_DEBUG,”(%t) handle_input ::”));

char* data= new char[13];

　

//Check if peer aborted the connection

if(peer().recv_n(data,12)==0)

{

printf(”Peer probably aborted connection”);

return -1; //de-register from the Reactor.

}

　

//Show what you got..

ACE_OS::printf(”<< %s\n”,data);

　

//keep yourself registered

return 0;

}

};

int main(int argc, char* argv[])

{

ACE_INET_Addr addr(10101);

ACE_DEBUG((LM_DEBUG,”Thread: (%t) main”));

//Prepare to accept connections

Acceptor myacceptor(addr,Reactor::instance());

// wait for something to happen.

while(1)

Reactor::instance()->handle_events();

　

return 0;

}