在本章里，我们描述怎样将反应堆用于对所有这些不同的事件类型进行多路分离。

　　于是用户（也就是，应用开发者）必须：

　　随后反应堆构架将自动地：

　　如上面所提到的，反应堆将事件处理器对象作为服务提供者使用。一旦反应堆成功地多路分离和分派了某事件，事件处理器对象就对它进行处理。因此，反应堆会在内部记住当特定类型的事件发生时，应该回调哪一个事件处理器对象。当应用在反应堆上登记它的处理器对象，以处理特定类型的事件时，反应堆会创建这种事件和相应的事件处理器的关联。

　　下面的简单例子可以帮助我们更好地理解这些步骤：

#include <signal.h>

#include ”ace/Reactor.h”

#include ”ace/Event_Handler.h”

//Create our subclass to handle the signal events

//that we wish to handle. Since we know that this particular

//event handler is going to be using signals we only overload the

//handle_signal method.

class MyEventHandler: public ACE_Event_Handler

{

int handle_signal(int signum, siginfo_t*,ucontext_t*)

{

switch(signum)

{

case SIGWINCH:

ACE_DEBUG((LM_DEBUG, ”You pressed SIGWINCH \n”));

break;

case SIGINT:

ACE_DEBUG((LM_DEBUG, ”You pressed SIGINT \n”));

break;

}

return 0;

}

};

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

{

//instantiate the handler

MyEventHandler *eh =new MyEventHandler;

//Register the handler asking to call back when either SIGWINCH

//or SIGINT signals occur. Note that in both the cases we asked the

//Reactor to callback the same Event_Handler i.e., MyEventHandler.

//This is the reason why we had to write a switch statement in the

//handle_signal() method above. Also note that the ACE_Reactor is

//being used as a Singleton object (Singleton pattern)

ACE_Reactor::instance()->register_handler(SIGWINCH,eh);

ACE_Reactor::instance()->register_handler(SIGINT,eh);

while(1)

//Start the reactors event loop

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

}

　　一旦所需的事件被处理后，可能就无需再让事件处理器登记在反应堆上。因而，反应堆提供了从它的内部分派表中拆除事件处理器的技术。一旦事件处理器被拆除，它就不再会被反应堆回调。

class MyEventHandler: public ACE_Event_Handler

{

public:

MyEventHandler(){//construct internal data members}

virtual int

handle_close(ACE_HANDLE handle, ACE_Reactor_Mask mask)

{

delete this; //commit suicide

}

~MyEventHandler(){//destroy internal data members}

private:

//internal data members

}

class MyEventHandler: public ACE_Event_Handler

{

public:

MyEventHandler(){//construct internal data members}

virtual int handle_close(ACE_HANDLE handle, ACE_Reactor_Mask mask)

{

dele te this; //commit suicide

}

private:

//Class must be allocated dynamically

~MyEventHandler(){//destroy internal data members}

};

　　在下面的几个部分，我们将演示怎样将反应堆用于处理各种类型的事件。

#include ”ace/Reactor.h”

#include ”ace/SOCK_Acceptor.h”

#define PORT_NO 19998

typedef ACE_SOCK_Acceptor Acceptor;

//forward declaration

class My_Accept_Handler;

class My_Input_Handler: public ACE_Event_Handler

{

public:

//Constructor

My_Input_Handler()

{

ACE_DEBUG((LM_DEBUG,”Constructor\n”);

}

//Called back to handle any input received

int handle_input(ACE_HANDLE)

{

//receive the data

peer().recv_n(data,12);

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

　

// do something with the input received.

// ...

//keep yourself registered with the reactor

return 0;

}

//Used by the reactor to determine the underlying handle

ACE_HANDLE get_handle()const

{

return this->peer_i().get_handle();

}

//Returns a reference to the underlying stream.

ACE_SOCK_Stream &peer_i()

{

return this->peer_;

}

private:

ACE_SOCK_Stream peer_;

char data [12];

};

class My_Accept_Handler: public ACE_Event_Handler

{

public:

//Constructor

My_Accept_Handler(ACE_Addr &addr)

{

this->open(addr);

}

//Open the peer_acceptor so it starts to ”listen”

//for incoming clients.

int open(ACE_Addr &addr)

{

peer_acceptor.open(addr);

return 0;

}

//Overload the handle input method

int handle_input(ACE_HANDLE handle)

{

//Client has requested connection to server.

//Create a handler to handle the connection

My_Input_Handler *eh= new My_Input_Handler();

　

//Accept the connection ”into” the Event Handler

if (this->peer_acceptor.accept (eh->peer (), // stream

0, // remote address

0, // timeout

1) ==-1) //restart if interrupted

ACE_DEBUG((LM_ERROR,”Error in connection\n”));

　

ACE_DEBUG((LM_DEBUG,”Connection established\n”));

　

//Register the input event handler for reading

ACE_Reactor::instance()->

register_handler(eh,ACE_Event_Handler::READ_MASK);

　

//Unregister as the acceptor is not expecting new clients

return -1;

}

//Used by the reactor to determine the underlying handle

ACE_HANDLE get_handle(void) const

{

return this->peer_acceptor.get_handle();

}

private:

Acceptor peer_acceptor;

};

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

{

//Create an address on which to receive connections

ACE_INET_Addr addr(PORT_NO);

//Create the Accept Handler which automatically begins to “listen”

//for client requests for connections

My_Accept_Handler *eh=new My_Accept_Handler(addr);

//Register the reactor to call back when incoming client connects

ACE_Reactor::instance()->register_handler(eh,

ACE_Event_Handler::ACCEPT_MASK);

　

//Start the event loop

while(1)

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

}

　　下面的例子演示怎样与反应堆一起使用定时器。

#include ”test_config.h”

#include ”ace/Timer_Queue.h”

#include ”ace/Reactor.h”

#define NUMBER_TIMERS 10

static int done = 0;

static int count = 0;

class Time_Handler : public ACE_Event_Handler

{

public:

//Method which is called back by the Reactor when timeout occurs.

virtual int handle_timeout (const ACE_Time_Value &tv,

const void *arg)

{

long current_count = long (arg);

ACE_ASSERT (current_count == count);

ACE_DEBUG ((LM_DEBUG, ”%d: Timer #%d timed out at %d!\n”,

count, current_count, tv.sec()));

　

//Increment count

count ++;

　

//Make sure assertion doesn’t fail for missing 5th timer.

if (count ==5)

count++;

　

//If all timers done then set done flag

if (current_count == NUMBER_TIMERS - 1)

done = 1;

　

//Keep yourself registered with the Reactor.

return 0;

}

};

int main (int, char *[])

{

ACE_Reactor reactor;

Time_Handler *th=new Time_Handler;

int timer_id[NUMBER_TIMERS];

int i;

for (i = 0; i < NUMBER_TIMERS; i++)

timer_id[i] = reactor.schedule_timer (th,

(const void *) i, // argument sent to handle_timeout()

ACE_Time_Value (2 * i + 1)); //set timer to go off with delay

　

//Cancel the fifth timer before it goes off

reactor.cancel_timer(timer_id[5]);//Timer ID of timer to be removed

while (!done)

reactor.handle_events ();

　

return 0;

}

　　不同的环境可能需要不同的调度和取消定时器的方法。在下面的任一条件为真时，实现定时器的算法的性能就会成为一个问题：

　　下面的例子将进一步阐释这些概念：

#include ”ace/Reactor.h”

#include ”ace/Event_Handler.h”

#include ”ace/Synch_T.h”

#include ”ace/Thread_Manager.h”

#define WAIT_TIME 1

#define SLEEP_TIME 2

class My_Handler: public ACE_Event_Handler

{

public:

//Start the event handling process.

My_Handler()

{

ACE_DEBUG((LM_DEBUG,”Event Handler created\n”));

ACE_Reactor::instance()->max_notify_iterations(5);

return 0;

}

//Perform the notifications i.e., notify the reactor 10 times

void perform_notifications()

{

for(int i=0;i<10;i++)

ACE_Reactor::instance()->

notify(this,ACE_Event_Handler::READ_MASK);

}

//The actual handler which in this case will handle the notifications

int handle_input(int)

{

ACE_DEBUG((LM_DEBUG,”Got notification # %d\n”,no));

no++;

return 0;

}

private:

static int no;

};

//Static members

int My_Handler::no=1;

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

{

ACE_DEBUG((LM_DEBUG,”Starting test \n”));

//Instantiating the handler

My_Handler handler;

//The done flag is set to not done yet.

int done=0;

while(1)

{

//After WAIT_TIME the handle_events will fall through if no events

//arrive.

ACE_Reactor::instance()->handle_events(ACE_Time_Value(WAIT_TIME));

if(!done)

{

handler.perform_notifications();

done=1;

}

　

sleep(SLEEP_TIME);

｝

｝

　　上面的例子过于简单，也非常不实际，因为通知发生的线程和反应堆所在的线程是同一线程。更为实际的例子是：事件发生在另一线程中，并将这些事件通知反应堆线程。下面所演示的是同一个例子，不过是由不同的线程来执行通知：

#include ”ace/Reactor.h”

#include ”ace/Event_Handler.h”

#include ”ace/Synch_T.h”

#include ”ace/Thread_Manager.h”

class My_Handler: public ACE_Event_Handler

{

public:

//Start the event handling process.

My_Handler()

{

ACE_DEBUG((LM_DEBUG,”Got open\n”));

activate_threads();

ACE_Reactor::instance()->max_notify_iterations(5);

return 0;

}

//Spawn a separate thread so that it notifies the reactor

void activate_threads()

{

ACE_Thread_Manager::instance()

->spawn((ACE_THR_FUNC)svc_start,(void*)this);

}

//Notify the Reactor 10 times.

void svc()

{

for(int i=0;i<10;i++)

ACE_Reactor::instance()->

notify(this, ACE_Event_Handler::READ_MASK);

}

//The actual handler which in this case will handle the notifications

int handle_input(int)

{

ACE_DEBUG((LM_DEBUG, ”Got notification # %d\n”, no));

no++;

return 0;

}

//The entry point for the new thread that is to be created.

static int svc_start(void* arg);

private:

static int no;

};

//Static members

int My_Handler::no=1;

int My_Handler::svc_start(void* arg)

{

My_Handler *eh= (My_Handler*)arg;

eh->svc();

return -1; //de-register from the reactor

}

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

{

ACE_DEBUG((LM_DEBUG,”Starting test \n”));

My_Handler handler;

while(1)

{

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

sleep(3);

}

}